From da33cad81d33bc33bd3f8196b70094cfff80f5ca Mon Sep 17 00:00:00 2001
From: root <root@localhost>
Date: Wed, 30 Sep 2026 00:24:21 +0800
Subject: [PATCH] =?UTF-8?q?feat(worker):=20IO=20Worker=20=E5=88=86?=
 =?UTF-8?q?=E9=81=93(WK-1)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

io worker 进程内按任务声明的 lane 分文件 / 网络 / 维护三条道,各自并发(缺省 2/4/1)、
各自排队;维护道在文件道或网络道有排队/运行时让路(连续 10 分钟放行一个),调度器内联的
清理/回填/保留期裁剪也走维护道。并发在 设置 → 自动化 → 高级 热生效,env 覆盖;旧
WORKER_CONCURRENCY 映射到网络道。运行状态页按道显示运行中/并发/排队。归属表写进
docs/runbook/dev.md,ci 与注册表逐项核对。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |   8 +
 apps/server/src/config.ts                     |  14 +-
 apps/server/src/context.ts                    |   4 +
 apps/server/src/main.ts                       |   8 +-
 .../modules/acquisition-organize/handlers.ts  |   5 +
 apps/server/src/modules/agent/tasks.ts        |   2 +
 .../server/src/modules/auth-settings/index.ts |   1 +
 apps/server/src/modules/cache/index.ts        |   3 +
 apps/server/src/modules/file-manager/index.ts |   2 +
 apps/server/src/modules/pi/index.ts           |   1 +
 apps/server/src/modules/schedules/registry.ts |   6 +
 .../server/src/modules/schedules/scheduler.ts |  13 +-
 .../src/modules/sources-identity/index.ts     |   1 +
 .../src/modules/sources-identity/tasks.ts     |   7 +
 apps/server/src/modules/system/health.ts      |  33 +-
 apps/server/src/modules/system/heartbeat.ts   |   3 +-
 apps/server/src/modules/system/index.ts       |   2 +
 apps/server/src/worker/index.ts               | 160 ++++++---
 apps/server/src/worker/lanes.ts               | 271 +++++++++++++++
 .../test/auth-settings/admin-api.test.ts      |   8 +-
 apps/server/test/invoke-limits.test.ts        |  28 +-
 apps/server/test/schedules/schedules.test.ts  |  53 +++
 apps/server/test/system/system.test.ts        |   2 +
 apps/server/test/worker-lanes.test.ts         | 319 ++++++++++++++++++
 .../app/components/settings/automation.tsx    |  36 +-
 apps/web/app/routes/console/status.tsx        |  28 +-
 docker-compose.yml                            |   9 +-
 docker/compose.env.example                    |   8 +-
 docs/runbook/deploy.md                        |   3 +-
 docs/runbook/dev.md                           |  64 +++-
 packages/contracts/src/ops.ts                 |  30 +-
 packages/contracts/src/settings.ts            |  34 +-
 packages/contracts/test/schemas.test.ts       |  10 +-
 packages/tasks/src/definitions.ts             |   9 +
 packages/tasks/src/index.ts                   |  10 +-
 packages/tasks/src/openworkflow-engine.ts     | 103 +++++-
 packages/tasks/src/registry.ts                |  19 ++
 packages/tasks/src/types.ts                   |  28 +-
 38 files changed, 1240 insertions(+), 105 deletions(-)
 create mode 100644 apps/server/src/worker/lanes.ts
 create mode 100644 apps/server/test/worker-lanes.test.ts

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 600ca151..0b0c1598 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,14 @@
 
 ## [Unreleased]
 
+### 新增
+- IO Worker 分道(WK-1):io worker 进程内分文件 / 网络 / 维护三条道,各自并发(缺省 2 / 4 / 1)、各自排队,文件道占满不挡网络任务;维护道在文件道或网络道有排队 / 运行时让路(连续让路 10 分钟放行一个),调度器里内联的清理、回填、保留期裁剪也走维护道。任务在 `defineTask({ lane })` 显式声明,归属表见 `docs/runbook/dev.md`「IO Worker 分道」,ci 逐项核对。
+- 设置 → 自动化 → 高级 →「后台任务并发」:三条道的并发,保存后几秒内生效,不用重启;环境变量 `WORKER_FILE_CONCURRENCY` / `WORKER_NETWORK_CONCURRENCY` / `WORKER_MAINTENANCE_CONCURRENCY` 设了以变量为准。
+- 运行状态页进程表按道显示「运行中/并发 · 排队」与维护道让路状态;心跳照常记录事件循环延迟(将来拆进程的判据)。
+
+### 变更
+- `WORKER_CONCURRENCY` 从 io 组总并发改为映射到网络道;compose 不再给它缺省值(缺省走设置页)。`worker` / `worker:agent` / `all` 角色与镜像入口不变。
+
 ## [0.3.0] - 2026-09-29
 
 ### 新增
diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts
index 0c1a86c3..17ffd104 100644
--- a/apps/server/src/config.ts
+++ b/apps/server/src/config.ts
@@ -4,6 +4,9 @@ import { z } from "zod";
 /** compose 里 `${X:-}` 展开成空串:按未设置处理 */
 const emptyToUndef = (v: unknown) => (v === "" ? undefined : v);
 
+/** 可选正整数环境变量(空串 = 未设置) */
+const optionalInt = (max: number) => z.preprocess(emptyToUndef, z.coerce.number().int().min(1).max(max).optional());
+
 const LogLevel = z.enum(["fatal", "error", "warn", "info", "debug", "trace", "silent"]);
 
 export const ConfigSchema = z.object({
@@ -29,8 +32,15 @@ export const ConfigSchema = z.object({
   DEV_ADMIN_KEY: z.string().min(8).optional(),
   /** worker 心跳间隔 */
   WORKER_HEARTBEAT_MS: z.coerce.number().int().min(1000).default(30_000),
-  /** io 组 worker(除 runAgent/resolveDecision 外的全部任务)并发槽数 */
-  WORKER_CONCURRENCY: z.coerce.number().int().min(1).max(64).default(4),
+  /**
+   * 旧变量(WK-1 前的 io 组总并发):现在映射到网络道(WORKER_NETWORK_CONCURRENCY 未设时);
+   * 不设 = 用设置页的值。空串视为未设置(compose 里 `${WORKER_CONCURRENCY:-}`)。
+   */
+  WORKER_CONCURRENCY: optionalInt(64),
+  /** io worker 三条道的并发(WK-1);设置了就覆盖 设置 → 自动化 → 高级 里的值 */
+  WORKER_FILE_CONCURRENCY: optionalInt(64),
+  WORKER_NETWORK_CONCURRENCY: optionalInt(64),
+  WORKER_MAINTENANCE_CONCURRENCY: optionalInt(64),
   /** agent 组 worker(runAgent/resolveDecision:整段会话占一槽数分钟)并发槽数 */
   WORKER_AGENT_CONCURRENCY: z.coerce.number().int().min(1).max(16).default(2),
   /** 手动整理 dryRun / 按路径整理时递归遍历的文件数上限(超出截断并标记 truncated) */
diff --git a/apps/server/src/context.ts b/apps/server/src/context.ts
index ef49ce1a..d895fea1 100644
--- a/apps/server/src/context.ts
+++ b/apps/server/src/context.ts
@@ -41,6 +41,10 @@ export class DeferredTaskEngine implements TaskEngine {
   get: TaskEngine["get"] = (...a) => this.e.get(...a);
   list: TaskEngine["list"] = (...a) => this.e.list(...a);
   startWorker: TaskEngine["startWorker"] = (...a) => this.e.startWorker(...a);
+  queueCounts: NonNullable<TaskEngine["queueCounts"]> = async (...a) => {
+    const q = this.e.queueCounts;
+    return q ? q.apply(this.e, a) : a[0].map(() => ({ queued: 0, running: 0 }));
+  };
   stop: TaskEngine["stop"] = async () => {
     await this.inner?.stop();
   };
diff --git a/apps/server/src/main.ts b/apps/server/src/main.ts
index 4cc6008d..cf7065f6 100644
--- a/apps/server/src/main.ts
+++ b/apps/server/src/main.ts
@@ -3,7 +3,7 @@ import denoJson from "../deno.json" with { type: "json" };
 import type { Config } from "./config.ts";
 import type { MappingExport } from "./ops.ts";
 
-// api / worker / worker:agent / all 是常驻角色(worker = io 组:除 runAgent/resolveDecision 外的任务 + outbox 中继 +
+// api / worker / worker:agent / all 是常驻角色(worker = io 组:文件 / 网络 / 维护三条道 + outbox 中继 +
 // 模块后台循环;worker:agent = 只跑 Agent 会话任务;all = api + 两组 worker 同进程);其余是容器入口 / 备份恢复脚本用的一次性子命令:
 //   migrate | preflight | probe
 //   export-mappings --out <file.json> [--compare <基准清单.json>]   映射清单导出(可与基准比行数)
@@ -77,6 +77,7 @@ const { SERVER_MODULES } = await import("./modules/index.ts");
 const { diffCounts, enableReadOnlyReconcile, exportMappings, writeMappingExport } = await import("./ops.ts");
 const { auditStorageWarnings, checkStorage, loadMountedPaths, runPreflight } = await import("./preflight.ts");
 const { startWorker } = await import("./worker/index.ts");
+const { laneOverridesFromEnv } = await import("./worker/lanes.ts");
 
 let config: Config;
 try {
@@ -253,9 +254,10 @@ if (role === "worker" || role === "worker:agent" || role === "all") {
       modules: SERVER_MODULES,
       setRelay: outbox.setRelay,
       group: role === "all" ? "split" : role === "worker" ? "io" : "agent",
-      taskNames: booted.tasks.names(),
-      concurrency: config.WORKER_CONCURRENCY,
+      tasks: booted.tasks.all(),
       agentConcurrency: config.WORKER_AGENT_CONCURRENCY,
+      // 三条道的并发读 设置 → 自动化 → 高级(热生效);环境变量设了就覆盖(旧 WORKER_CONCURRENCY → 网络道)
+      lanes: { overrides: laneOverridesFromEnv(config) },
     },
   });
   stops.push(worker.stop);
diff --git a/apps/server/src/modules/acquisition-organize/handlers.ts b/apps/server/src/modules/acquisition-organize/handlers.ts
index e6344625..73c2c44b 100644
--- a/apps/server/src/modules/acquisition-organize/handlers.ts
+++ b/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) => {
diff --git a/apps/server/src/modules/agent/tasks.ts b/apps/server/src/modules/agent/tasks.ts
index 55d8508b..5663a31a 100644
--- a/apps/server/src/modules/agent/tasks.ts
+++ b/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) => {
diff --git a/apps/server/src/modules/auth-settings/index.ts b/apps/server/src/modules/auth-settings/index.ts
index 4995f91f..a20f3fdb 100644
--- a/apps/server/src/modules/auth-settings/index.ts
+++ b/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 天前失效的会话",
diff --git a/apps/server/src/modules/cache/index.ts b/apps/server/src/modules/cache/index.ts
index 7e2ae53f..8b1a40ad 100644
--- a/apps/server/src/modules/cache/index.ts
+++ b/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: "海报缓存超过容量上限时，删掉最久没用的",
diff --git a/apps/server/src/modules/file-manager/index.ts b/apps/server/src/modules/file-manager/index.ts
index 98b7171a..e8d416bf 100644
--- a/apps/server/src/modules/file-manager/index.ts
+++ b/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) =>
diff --git a/apps/server/src/modules/pi/index.ts b/apps/server/src/modules/pi/index.ts
index 401dadd9..2616fc79 100644
--- a/apps/server/src/modules/pi/index.ts
+++ b/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) =>
diff --git a/apps/server/src/modules/schedules/registry.ts b/apps/server/src/modules/schedules/registry.ts
index f65e43d8..9a203b79 100644
--- a/apps/server/src/modules/schedules/registry.ts
+++ b/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;
 }
diff --git a/apps/server/src/modules/schedules/scheduler.ts b/apps/server/src/modules/schedules/scheduler.ts
index a24289de..710735c7 100644
--- a/apps/server/src/modules/schedules/scheduler.ts
+++ b/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);
     }
diff --git a/apps/server/src/modules/sources-identity/index.ts b/apps/server/src/modules/sources-identity/index.ts
index 6fd1bd19..797b0997 100644
--- a/apps/server/src/modules/sources-identity/index.ts
+++ b/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: "按语言设置重新拉一遍作品的标题、简介和每集信息；跑完自动停用，可手动再执行",
diff --git a/apps/server/src/modules/sources-identity/tasks.ts b/apps/server/src/modules/sources-identity/tasks.ts
index 6aaa9dcb..8e7babbb 100644
--- a/apps/server/src/modules/sources-identity/tasks.ts
+++ b/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 },
diff --git a/apps/server/src/modules/system/health.ts b/apps/server/src/modules/system/health.ts
index 97acf692..207e5639 100644
--- a/apps/server/src/modules/system/health.ts
+++ b/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,
diff --git a/apps/server/src/modules/system/heartbeat.ts b/apps/server/src/modules/system/heartbeat.ts
index 09d1b0c0..00cf5f42 100644
--- a/apps/server/src/modules/system/heartbeat.ts
+++ b/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>;
diff --git a/apps/server/src/modules/system/index.ts b/apps/server/src/modules/system/index.ts
index 8af99ac2..1fbbd4f0 100644
--- a/apps/server/src/modules/system/index.ts
+++ b/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: "删除过期的进程状态记录",
diff --git a/apps/server/src/worker/index.ts b/apps/server/src/worker/index.ts
index 88a373cb..14761bcc 100644
--- a/apps/server/src/worker/index.ts
+++ b/apps/server/src/worker/index.ts
@@ -1,10 +1,19 @@
 import { EventType } from "@umt/contracts";
-import { OutboxRelay } from "@umt/tasks";
+import { OutboxRelay, type WorkerHandle } from "@umt/tasks";
 import type { Logger } from "pino";
 import { collectOutboxRoutes, tasksLogSink } from "../context.ts";
 import { type EventBusPort, sseConnectionCount } from "../http/sse.ts";
 import { type HeartbeatWriter, startHeartbeatWriter } from "../modules/system/heartbeat.ts";
 import type { ServerContext, ServerModule } from "../modules/types.ts";
+import {
+  groupTasksByLane,
+  IO_LANES,
+  type IoLane,
+  LaneController,
+  type LaneOverrides,
+  type LaneTask,
+  setLaneController,
+} from "./lanes.ts";
 
 /**
  * outbox → SSE 桥:这些前缀的 outbox 事件在中继投递成功后镜像到实时总线(payload 原样,带业务 id),
@@ -16,18 +25,20 @@ export const SSE_MIRRORED_OUTBOX_TYPES: readonly string[] = Object.values(EventT
   (t) => SSE_MIRROR_PREFIXES.some((p) => t.startsWith(p)) && !SSE_MIRROR_EXCLUDED.has(t),
 );
 
-/** agent 组任务:整段会话占一槽数分钟,与文件复制等 io 任务分池 */
-export const AGENT_TASKS = ["runAgent", "resolveDecision"] as const;
+/** agent 组任务:整段会话占一槽数分钟,与文件复制等 io 任务分池(以任务声明的 lane = "agent" 为准,这是兜底名单) */
+export { AGENT_TASKS } from "./lanes.ts";
 
 /**
  * worker 分组:
  *   all   — 单个引擎 worker 执行全部任务(测试 / 省略分组时的兼容行为)
- *   io    — 除 AGENT_TASKS 外的全部任务 + OutboxRelay + 模块后台循环
- *   agent — 只执行 AGENT_TASKS;不跑 relay 与模块循环
- *   split — 同一进程里 io + agent 两个引擎 worker(--role all)
+ *   io    — 除 agent 道外的全部任务,按 lane 分文件 / 网络 / 维护三个引擎 worker + OutboxRelay + 模块后台循环
+ *   agent — 只执行 agent 道任务;不跑 relay 与模块循环
+ *   split — 同一进程里 io 三条道 + agent 共四个引擎 worker(--role all)
  */
 export type WorkerGroup = "all" | "io" | "agent" | "split";
 
+export type PoolLabel = "all" | IoLane | "agent";
+
 export interface WorkerDeps {
   logger: Logger;
   heartbeatMs: number;
@@ -40,14 +51,21 @@ export interface WorkerDeps {
     /** 设置 outbox 中继实例到 ctx.outbox.relay */
     setRelay?: (relay: OutboxRelay) => void;
     outboxIntervalMs?: number;
-    /** io 组(或 all 组)并发 */
+    /** all 组(单池)并发 */
     concurrency?: number;
     /** agent 组并发 */
     agentConcurrency?: number;
     /** 缺省 all */
     group?: WorkerGroup;
-    /** 注册表里的全部任务名(io 组 = 全部 − AGENT_TASKS);group 为 io/split 时必填 */
-    taskNames?: readonly string[];
+    /** 注册表里的全部任务(名称 + lane);group 为 io/agent/split 时必填 */
+    tasks?: readonly LaneTask[];
+    /** 分道:环境变量覆盖与测试用的节奏参数 */
+    lanes?: {
+      overrides?: LaneOverrides;
+      refreshMs?: number;
+      busyCacheMs?: number;
+      maxYieldMs?: number;
+    };
   };
   /** 事件循环延迟探针:采样间隔与告警阈值(测试可调小) */
   lagProbe?: { intervalMs?: number; warnMs?: number };
@@ -63,31 +81,38 @@ export interface WorkerStats {
 export interface RunningWorker {
   stop(): Promise<void>;
   stats(): WorkerStats;
+  /** io / split 组的分道控制器 */
+  lanes: LaneController | null;
 }
 
-/** 按分组拆出引擎 worker 的任务选择(versions 为任务名,匹配该任务全部版本) */
+export interface PoolPlan {
+  label: PoolLabel;
+  /** 任务名(匹配该任务全部版本);all 池省略 */
+  versions?: string[];
+  concurrency?: number;
+}
+
+/** 按分组拆出引擎 worker 的任务选择;io 组按 lane 拆三条道(空道不起) */
 export function workerPlan(
   group: WorkerGroup,
-  taskNames: readonly string[],
-  concurrency: { io?: number; agent?: number },
-): { label: "all" | "io" | "agent"; versions?: string[]; concurrency?: number }[] {
-  const agentNames = taskNames.filter((n) => (AGENT_TASKS as readonly string[]).includes(n));
-  const ioNames = taskNames.filter((n) => !(AGENT_TASKS as readonly string[]).includes(n));
-  const io = { label: "io" as const, versions: ioNames, ...(concurrency.io ? { concurrency: concurrency.io } : {}) };
-  const agent = {
-    label: "agent" as const,
-    versions: agentNames,
-    ...(concurrency.agent ? { concurrency: concurrency.agent } : {}),
-  };
+  tasks: readonly LaneTask[],
+  concurrency: { all?: number; agent?: number } & LaneOverrides = {},
+): PoolPlan[] {
+  const { byLane } = groupTasksByLane(tasks);
+  const pool = (label: IoLane | "agent"): PoolPlan[] =>
+    byLane[label].length
+      ? [{ label, versions: byLane[label], ...(concurrency[label] ? { concurrency: concurrency[label] } : {}) }]
+      : [];
+  const io = IO_LANES.flatMap((l) => pool(l));
   switch (group) {
     case "all":
-      return [{ label: "all", ...(concurrency.io ? { concurrency: concurrency.io } : {}) }];
+      return [{ label: "all", ...(concurrency.all ? { concurrency: concurrency.all } : {}) }];
     case "io":
-      return ioNames.length ? [io] : [];
+      return io;
     case "agent":
-      return agentNames.length ? [agent] : [];
+      return pool("agent");
     case "split":
-      return [...(ioNames.length ? [io] : []), ...(agentNames.length ? [agent] : [])];
+      return [...io, ...pool("agent")];
   }
 }
 
@@ -134,18 +159,50 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe
 
   const started: ServerModule[] = [];
   let relay: OutboxRelay | null = null;
-  const plan: ReturnType<typeof workerPlan> = [];
+  let lanes: LaneController | null = null;
+  const plan: PoolPlan[] = [];
+  const handles = new Map<PoolLabel, WorkerHandle>();
   if (runtime) {
     const { ctx } = runtime;
-    if (group !== "all" && !runtime.taskNames) throw new Error(`worker 分组 ${group} 需要 taskNames`);
-    plan.push(
-      ...workerPlan(group, runtime.taskNames ?? [], { io: runtime.concurrency, agent: runtime.agentConcurrency }),
+    if (group !== "all" && !runtime.tasks) throw new Error(`worker 分组 ${group} 需要 tasks`);
+    const tasks = runtime.tasks ?? [];
+    plan.push(...workerPlan(group, tasks, { all: runtime.concurrency, agent: runtime.agentConcurrency }));
+    const laned = plan.filter((p): p is PoolPlan & { label: IoLane } =>
+      (IO_LANES as readonly string[]).includes(p.label),
     );
-    for (const p of plan) {
-      await ctx.engine.startWorker({
-        ...(p.versions ? { versions: p.versions } : {}),
-        ...(p.concurrency ? { concurrency: p.concurrency } : {}),
+    if (laned.length) {
+      const { byLane, missing } = groupTasksByLane(tasks);
+      if (missing.length)
+        log.warn({ tasks: missing }, "这些任务没有声明 lane，按缺省归入网络道（请在 defineTask 里补 lane）");
+      lanes = new LaneController({
+        engine: ctx.engine,
+        logger,
+        lanes: { file: byLane.file, network: byLane.network, maintenance: byLane.maintenance },
+        readSettings: async () => (await ctx.settings.get("automation")).workerLanes,
+        ...(runtime.lanes?.overrides ? { overrides: runtime.lanes.overrides } : {}),
+        ...(runtime.lanes?.refreshMs ? { refreshMs: runtime.lanes.refreshMs } : {}),
+        ...(runtime.lanes?.busyCacheMs !== undefined ? { busyCacheMs: runtime.lanes.busyCacheMs } : {}),
+        ...(runtime.lanes?.maxYieldMs ? { maxYieldMs: runtime.lanes.maxYieldMs } : {}),
       });
+      await lanes.start();
+      setLaneController(ctx, lanes);
+    }
+    for (const p of plan) {
+      const lane = (IO_LANES as readonly string[]).includes(p.label) ? (p.label as IoLane) : null;
+      const handle =
+        lane && lanes
+          ? await ctx.engine.startWorker({
+              versions: p.versions ?? [],
+              concurrency: lanes.slots(lane),
+              maxActive: () => lanes?.maxActive(lane) ?? 0,
+              ...(lane === "maintenance" ? { claimGate: () => lanes?.maintenanceMayStart() ?? true } : {}),
+            })
+          : await ctx.engine.startWorker({
+              ...(p.versions ? { versions: p.versions } : {}),
+              ...(p.concurrency ? { concurrency: p.concurrency } : {}),
+            });
+      if (handle) handles.set(p.label, handle);
+      if (lane) lanes?.attach(lane, handle || undefined);
     }
   }
   // outbox 中继与模块后台循环只在 io 组(agent 组进程只执行会话任务)
@@ -170,12 +227,14 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe
     }
   }
 
-  // 进程与健康页:每 15s 把心跳/延迟/并发/运行中 run 数写进 process_heartbeat(写失败不影响 worker)
+  // 进程与健康页:每 15s 把心跳/延迟/并发/运行中 run 数写进 process_heartbeat(写失败不影响 worker);
+  // io 组另写各道的并发、本进程运行中、全库排队(extra.lanes),事件循环延迟照常记录(将来拆进程的判据)
   let hbWriter: HeartbeatWriter | null = null;
   if (runtime) {
     const { ctx } = runtime;
     const groupTasks = new Set(plan.flatMap((p) => p.versions ?? []));
     const countAll = plan.some((p) => !p.versions);
+    const agentPool = plan.find((p) => p.label === "agent");
     hbWriter = startHeartbeatWriter({
       db: ctx.db,
       logger,
@@ -183,16 +242,27 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe
       sample: async () => {
         const running = await ctx.engine.list({ status: "running", limit: 500 }).catch(() => null);
         const s = probe.stats();
+        const laneStats = lanes ? await lanes.snapshot() : null;
+        const concurrency: Record<string, number> = {};
+        if (laneStats) {
+          for (const l of laneStats) concurrency[l.lane] = l.concurrency;
+          concurrency.io = laneStats.reduce((a, l) => a + l.concurrency, 0);
+        } else if (plan.some((p) => p.label === "all")) {
+          concurrency.io = runtime.concurrency ?? 4;
+        }
+        if (agentPool) concurrency.agent = handles.get("agent")?.concurrency ?? runtime.agentConcurrency ?? 4;
         return {
-          concurrency: {
-            ...(plan.some((p) => p.label !== "agent") ? { io: runtime.concurrency ?? 4 } : {}),
-            ...(plan.some((p) => p.label === "agent") ? { agent: runtime.agentConcurrency ?? 4 } : {}),
-          },
+          concurrency,
           runningRuns: running
             ? running.filter((r) => r.status === "running" && (countAll || groupTasks.has(r.taskName))).length
             : null,
           sseConnections: group === "split" ? sseConnectionCount() : null,
-          extra: { group, lagWarnings: s.loopLagWarnings, probeMaxLagMs: s.maxLoopLagMs },
+          extra: {
+            group,
+            lagWarnings: s.loopLagWarnings,
+            probeMaxLagMs: s.maxLoopLagMs,
+            ...(laneStats ? { lanes: laneStats } : {}),
+          },
         };
       },
     });
@@ -202,7 +272,14 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe
     {
       heartbeatMs,
       modules: started.map((m) => m.name),
-      pools: plan.map((p) => ({ group: p.label, concurrency: p.concurrency ?? 4, tasks: p.versions?.length ?? "all" })),
+      pools: plan.map((p) => ({
+        group: p.label,
+        concurrency:
+          lanes && (IO_LANES as readonly string[]).includes(p.label)
+            ? lanes.limit(p.label as IoLane)
+            : (p.concurrency ?? 4),
+        tasks: p.versions?.length ?? "all",
+      })),
     },
     "worker started",
   );
@@ -210,9 +287,12 @@ export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe
   const timer = setInterval(beat, heartbeatMs);
   return {
     stats: () => probe.stats(),
+    lanes,
     async stop() {
       clearInterval(timer);
       probe.stop();
+      lanes?.stop();
+      if (runtime) setLaneController(runtime.ctx, null);
       await hbWriter?.stop();
       for (const m of started.reverse()) {
         try {
diff --git a/apps/server/src/worker/lanes.ts b/apps/server/src/worker/lanes.ts
new file mode 100644
index 00000000..5d47c90b
--- /dev/null
+++ b/apps/server/src/worker/lanes.ts
@@ -0,0 +1,271 @@
+import { DEFAULT_WORKER_LANES, WORKER_LANE_MAX, type WorkerLaneSettings } from "@umt/contracts";
+import { laneOf, type QueueCounts, type TaskDefinition, type TaskLane, type WorkerHandle } from "@umt/tasks";
+import type { Logger } from "pino";
+import type { Config } from "../config.ts";
+
+// IO Worker 分道(WK-1,docs/runbook/dev.md「IO Worker 分道」):
+// io worker 进程内按任务声明的 lane 起三个引擎 worker(文件 / 网络 / 维护),各自独立并发;
+// 维护道"空闲时跑":文件道或网络道(全库)有排队或运行中的 run 时,维护道暂不领取新活(已在跑的不打断),
+// 连续让路超过 MAINTENANCE_MAX_YIELD_MS 放行一个,防止网络道长期有活时维护永远跑不上。
+// 调度器里内联执行的维护类定时任务(ScheduledJobDef.lane = "maintenance")与维护道共用并发与让路规则。
+
+export const IO_LANES = ["file", "network", "maintenance"] as const;
+export type IoLane = (typeof IO_LANES)[number];
+
+export const LANE_LABEL: Record<IoLane, string> = { file: "文件", network: "网络", maintenance: "维护" };
+
+/** 维护道最长连续让路时长,超过即放行一个 */
+export const MAINTENANCE_MAX_YIELD_MS = 10 * 60_000;
+
+export type LaneOverrides = Partial<Record<IoLane, number>>;
+
+/** 环境变量覆盖:旧 WORKER_CONCURRENCY 映射到网络道(WORKER_NETWORK_CONCURRENCY 优先) */
+export function laneOverridesFromEnv(
+  config: Pick<
+    Config,
+    "WORKER_CONCURRENCY" | "WORKER_FILE_CONCURRENCY" | "WORKER_NETWORK_CONCURRENCY" | "WORKER_MAINTENANCE_CONCURRENCY"
+  >,
+): LaneOverrides {
+  const network = config.WORKER_NETWORK_CONCURRENCY ?? config.WORKER_CONCURRENCY;
+  return {
+    ...(config.WORKER_FILE_CONCURRENCY ? { file: config.WORKER_FILE_CONCURRENCY } : {}),
+    ...(network ? { network } : {}),
+    ...(config.WORKER_MAINTENANCE_CONCURRENCY ? { maintenance: config.WORKER_MAINTENANCE_CONCURRENCY } : {}),
+  };
+}
+
+export interface LaneTask {
+  name: string;
+  lane?: TaskLane;
+}
+
+/** agent 会话任务:没声明 lane 时也归 agent(兼容 WK-1 前的固定名单) */
+export const AGENT_TASKS = ["runAgent", "resolveDecision"] as const;
+
+/**
+ * 按 lane 分组(同名多版本取第一次出现的声明);missing = 没声明 lane 的任务
+ * (AGENT_TASKS 归 agent,其余按缺省归网络道)。
+ */
+export function groupTasksByLane(tasks: readonly LaneTask[]): {
+  byLane: Record<TaskLane, string[]>;
+  missing: string[];
+} {
+  const byLane: Record<TaskLane, string[]> = { file: [], network: [], maintenance: [], agent: [] };
+  const missing: string[] = [];
+  const seen = new Set<string>();
+  for (const t of tasks) {
+    if (seen.has(t.name)) continue;
+    seen.add(t.name);
+    if (t.lane === undefined) missing.push(t.name);
+    const lane =
+      t.lane === undefined && (AGENT_TASKS as readonly string[]).includes(t.name)
+        ? "agent"
+        : laneOf(t as Pick<TaskDefinition, "lane">);
+    byLane[lane].push(t.name);
+  }
+  for (const k of Object.keys(byLane) as TaskLane[]) byLane[k].sort();
+  return { byLane, missing: missing.sort() };
+}
+
+export interface LaneStat {
+  lane: IoLane;
+  /** 当前生效的并发上限 */
+  concurrency: number;
+  /** 上限来源 */
+  source: "env" | "settings";
+  /** 本进程该道正在执行的 run 数(维护道含调度器内联的维护任务) */
+  running: number;
+  /** 全库该道已到点等待领取的 run 数;统计失败为 null */
+  queued: number | null;
+  /** 维护道:当前是否在给文件道 / 网络道让路 */
+  yielding?: boolean;
+}
+
+export interface LaneControllerOptions {
+  engine: { queueCounts?(groups: readonly (readonly string[])[]): Promise<QueueCounts[]> };
+  logger: Logger;
+  /** 各道的任务名 */
+  lanes: Record<IoLane, readonly string[]>;
+  /** 读设置页的并发(设置读取器自带缓存,settings.changed 事件立即失效) */
+  readSettings: () => Promise<WorkerLaneSettings>;
+  overrides?: LaneOverrides;
+  /** 重读设置的间隔,缺省 5s */
+  refreshMs?: number;
+  /** 忙闲判断缓存,缺省 1s */
+  busyCacheMs?: number;
+  maxYieldMs?: number;
+  now?: () => number;
+}
+
+export class LaneController {
+  private readonly limits: Record<IoLane, number>;
+  private readonly handles: Partial<Record<IoLane, WorkerHandle>> = {};
+  private inlineMaintenance = 0;
+  private yieldingSince: number | null = null;
+  private busyCache: { at: number; value: Promise<boolean> } | null = null;
+  private timer: ReturnType<typeof setInterval> | null = null;
+  private readonly log: Logger;
+  private readonly now: () => number;
+
+  constructor(private readonly o: LaneControllerOptions) {
+    this.log = o.logger.child({ component: "worker-lanes" });
+    this.now = o.now ?? (() => Date.now());
+    this.limits = { ...DEFAULT_WORKER_LANES, ...o.overrides };
+  }
+
+  /** 引擎 worker 的槽位数:设置页可调到 WORKER_LANE_MAX;环境变量更大时按环境变量开 */
+  slots(lane: IoLane): number {
+    return Math.max(WORKER_LANE_MAX, this.o.overrides?.[lane] ?? 0);
+  }
+
+  source(lane: IoLane): "env" | "settings" {
+    return this.o.overrides?.[lane] ? "env" : "settings";
+  }
+
+  limit(lane: IoLane): number {
+    return this.limits[lane];
+  }
+
+  tasks(lane: IoLane): readonly string[] {
+    return this.o.lanes[lane];
+  }
+
+  attach(lane: IoLane, handle: WorkerHandle | undefined): void {
+    if (handle) this.handles[lane] = handle;
+  }
+
+  /** 本进程该道正在执行的数 */
+  active(lane: IoLane): number {
+    const engine = this.handles[lane]?.active() ?? 0;
+    return lane === "maintenance" ? engine + this.inlineMaintenance : engine;
+  }
+
+  /** 引擎 worker 的动态上限;维护道扣掉调度器内联占用的 */
+  maxActive(lane: IoLane): number {
+    return lane === "maintenance" ? this.limits.maintenance - this.inlineMaintenance : this.limits[lane];
+  }
+
+  /** 重读设置 + 环境变量覆盖;变化时记日志 */
+  async refresh(): Promise<void> {
+    let fromSettings: WorkerLaneSettings = DEFAULT_WORKER_LANES;
+    try {
+      fromSettings = { ...DEFAULT_WORKER_LANES, ...(await this.o.readSettings()) };
+    } catch (err) {
+      this.log.debug({ err }, "读取分道并发设置失败，沿用上次的值");
+      return;
+    }
+    const changed: Record<string, number> = {};
+    for (const lane of IO_LANES) {
+      const next = Math.min(this.slots(lane), this.o.overrides?.[lane] ?? fromSettings[lane]);
+      if (next !== this.limits[lane]) changed[lane] = next;
+      this.limits[lane] = next;
+    }
+    if (Object.keys(changed).length) this.log.info({ changed, limits: { ...this.limits } }, "分道并发已更新");
+  }
+
+  start(): Promise<void> {
+    if (!this.timer) {
+      this.timer = setInterval(() => void this.refresh(), this.o.refreshMs ?? 5000);
+      try {
+        Deno.unrefTimer(this.timer);
+      } catch {
+        // 非 Deno 运行时
+      }
+    }
+    return this.refresh();
+  }
+
+  stop(): void {
+    if (this.timer) clearInterval(this.timer);
+    this.timer = null;
+  }
+
+  /** 文件道或网络道(全库)有排队或运行中的 run;统计失败按空闲处理(不让维护道因统计故障永远饿着) */
+  othersBusy(): Promise<boolean> {
+    const now = this.now();
+    if (!this.busyCache || now - this.busyCache.at >= (this.o.busyCacheMs ?? 1000)) {
+      const names = [...this.o.lanes.file, ...this.o.lanes.network];
+      const value = (async () => {
+        if (!this.o.engine.queueCounts || names.length === 0) return false;
+        try {
+          const [c] = await this.o.engine.queueCounts([names]);
+          return !!c && c.queued + c.running > 0;
+        } catch (err) {
+          this.log.debug({ err }, "分道忙闲统计失败");
+          return false;
+        }
+      })();
+      this.busyCache = { at: now, value };
+    }
+    return this.busyCache.value;
+  }
+
+  /** 维护道是否可以领取新活:其它两道空闲,或连续让路已超过上限(放行一个并重新计时) */
+  async maintenanceMayStart(): Promise<boolean> {
+    if (!(await this.othersBusy())) {
+      this.yieldingSince = null;
+      return true;
+    }
+    const now = this.now();
+    this.yieldingSince ??= now;
+    if (now - this.yieldingSince >= (this.o.maxYieldMs ?? MAINTENANCE_MAX_YIELD_MS)) {
+      this.log.warn({ yieldedMs: now - this.yieldingSince }, "维护道让路超时，放行一个维护任务");
+      this.yieldingSince = now;
+      return true;
+    }
+    return false;
+  }
+
+  yielding(): boolean {
+    return this.yieldingSince !== null;
+  }
+
+  /**
+   * 调度器内联维护任务进入维护道:占一个维护并发;force(手动立即执行)跳过让路与上限。
+   * 返回释放函数;不能进入时返回 null(任务保持到期状态,下一轮再试)。
+   */
+  async enterMaintenance(force = false): Promise<(() => void) | null> {
+    if (!force && this.active("maintenance") >= this.limits.maintenance) return null;
+    this.inlineMaintenance += 1;
+    if (!force && !(await this.maintenanceMayStart())) {
+      this.inlineMaintenance -= 1;
+      return null;
+    }
+    let released = false;
+    return () => {
+      if (released) return;
+      released = true;
+      this.inlineMaintenance -= 1;
+    };
+  }
+
+  /** 心跳 / 运行状态页用:各道并发、本进程运行中、全库排队 */
+  async snapshot(): Promise<LaneStat[]> {
+    let counts: (QueueCounts | undefined)[] = [];
+    try {
+      counts = (await this.o.engine.queueCounts?.(IO_LANES.map((l) => this.o.lanes[l]))) ?? [];
+    } catch (err) {
+      this.log.debug({ err }, "分道排队统计失败");
+    }
+    return IO_LANES.map((lane, i) => ({
+      lane,
+      concurrency: this.limits[lane],
+      source: this.source(lane),
+      running: this.active(lane),
+      queued: counts[i]?.queued ?? null,
+      ...(lane === "maintenance" ? { yielding: this.yielding() } : {}),
+    }));
+  }
+}
+
+// 同一 ServerContext 共用一个分道控制器(调度器据此给内联维护任务让路);没起 io worker 时为空
+const controllers = new WeakMap<object, LaneController>();
+
+export function setLaneController(ctx: object, c: LaneController | null): void {
+  if (c) controllers.set(ctx, c);
+  else controllers.delete(ctx);
+}
+
+export function laneControllerOf(ctx: object): LaneController | undefined {
+  return controllers.get(ctx);
+}
diff --git a/apps/server/test/auth-settings/admin-api.test.ts b/apps/server/test/auth-settings/admin-api.test.ts
index 984dac78..0d46bc73 100644
--- a/apps/server/test/auth-settings/admin-api.test.ts
+++ b/apps/server/test/auth-settings/admin-api.test.ts
@@ -75,7 +75,11 @@ describe.skipIf(!hasPg)("W2-A 管理 API(真实 PG)", () => {
   describe("设置", () => {
     it("GET 缺省值;PATCH 自动化总开关与只读对账;ctx.settings 缓存失效并广播 settings.changed", async () => {
       const g = await json(await admin("GET", "/settings"));
-      expect(g.data.automation).toEqual({ enabled: true, readOnlyReconcile: false });
+      expect(g.data.automation).toEqual({
+        enabled: true,
+        readOnlyReconcile: false,
+        workerLanes: { file: 2, network: 4, maintenance: 1 },
+      });
       expect((await h.ctx.settings.get("automation")).enabled).toBe(true);
       const events: string[] = [];
       const off = h.bus.subscribe((e) => events.push(e.type));
@@ -83,7 +87,7 @@ describe.skipIf(!hasPg)("W2-A 管理 API(真实 PG)", () => {
       expect(r.status).toBe(200);
       off();
       expect(events).toContain("settings.changed");
-      expect(await h.ctx.settings.get("automation")).toEqual({ enabled: false, readOnlyReconcile: true });
+      expect(await h.ctx.settings.get("automation")).toMatchObject({ enabled: false, readOnlyReconcile: true });
       // 其它分组不变
       expect((await json(await admin("GET", "/settings"))).data.cache.hardThresholdPercent).toBe(90);
     });
diff --git a/apps/server/test/invoke-limits.test.ts b/apps/server/test/invoke-limits.test.ts
index eb0a70cb..aa8cd4c9 100644
--- a/apps/server/test/invoke-limits.test.ts
+++ b/apps/server/test/invoke-limits.test.ts
@@ -71,19 +71,29 @@ describe("能力调用分级(HTTP 路径)", () => {
 });
 
 describe("worker 分组与事件循环探针", () => {
-  const names = ["evaluateAcquisition", "resolveDecision", "runAgent", "verifyAndImport"];
-  it("io / agent / split 的任务选择", () => {
-    expect(workerPlan("io", names, { io: 4 })).toEqual([
-      { label: "io", versions: ["evaluateAcquisition", "verifyAndImport"], concurrency: 4 },
+  const tasks = [
+    { name: "evaluateAcquisition", lane: "network" as const },
+    { name: "resolveDecision", lane: "agent" as const },
+    { name: "runAgent" },
+    { name: "verifyAndImport", lane: "file" as const },
+    { name: "purgeTrash", lane: "maintenance" as const },
+  ];
+  it("io / agent / split 的任务选择(io 按 lane 分三条道;没声明 lane 的 runAgent 兜底归 agent)", () => {
+    expect(workerPlan("io", tasks).map((p) => [p.label, p.versions])).toEqual([
+      ["file", ["verifyAndImport"]],
+      ["network", ["evaluateAcquisition"]],
+      ["maintenance", ["purgeTrash"]],
     ]);
-    expect(workerPlan("agent", names, { agent: 2 })).toEqual([
+    expect(workerPlan("agent", tasks, { agent: 2 })).toEqual([
       { label: "agent", versions: ["resolveDecision", "runAgent"], concurrency: 2 },
     ]);
-    expect(workerPlan("split", names, { io: 3, agent: 1 }).map((p) => [p.label, p.concurrency])).toEqual([
-      ["io", 3],
-      ["agent", 1],
+    expect(workerPlan("split", tasks, { agent: 1 }).map((p) => p.label)).toEqual([
+      "file",
+      "network",
+      "maintenance",
+      "agent",
     ]);
-    expect(workerPlan("all", names, {})).toEqual([{ label: "all" }]);
+    expect(workerPlan("all", tasks, {})).toEqual([{ label: "all" }]);
   });
 
   it("主线程被同步占住超过阈值 → 计数", async () => {
diff --git a/apps/server/test/schedules/schedules.test.ts b/apps/server/test/schedules/schedules.test.ts
index b49f4a0a..933ab4b2 100644
--- a/apps/server/test/schedules/schedules.test.ts
+++ b/apps/server/test/schedules/schedules.test.ts
@@ -12,6 +12,7 @@ import {
 } from "../../src/modules/schedules/registry.ts";
 import { Scheduler } from "../../src/modules/schedules/scheduler.ts";
 import type { ServerContext } from "../../src/modules/types.ts";
+import { LaneController, setLaneController } from "../../src/worker/lanes.ts";
 import { Browser, bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts";
 import { DEV_KEY, json } from "../helpers.ts";
 
@@ -94,6 +95,58 @@ describe.skipIf(!hasPg)("定时任务注册表(真实 PG)", () => {
     await h?.close();
   });
 
+  it("WK-1:lane=maintenance 的内联任务在文件道 / 网络道有活时让路(保持到期),空闲后执行;立即执行不让路", async () => {
+    let busy = true;
+    const logger = ctx.logger;
+    const lanes = new LaneController({
+      engine: { queueCounts: async (g) => g.map(() => ({ queued: busy ? 1 : 0, running: 0 })) },
+      logger,
+      lanes: { file: ["f"], network: ["n"], maintenance: [] },
+      readSettings: async () => ({ file: 2, network: 4, maintenance: 1 }),
+      busyCacheMs: 0,
+      maxYieldMs: 3_600_000,
+    });
+    await lanes.refresh();
+    setLaneController(ctx, lanes);
+    try {
+      registerScheduledJob(ctx, counting("test.lane_maint", { lane: "maintenance" }));
+      registerScheduledJob(ctx, counting("test.lane_light"));
+      const started = await sched.tick();
+      await sched.idle();
+      expect(started).toContain("test.lane_light");
+      expect(started).not.toContain("test.lane_maint");
+      expect(calls.get("test.lane_maint") ?? 0).toBe(0);
+      expect(lanes.yielding()).toBe(true);
+
+      // 立即执行:不让路
+      await h.db
+        .updateTable("scheduled_job")
+        .set({ run_requested_at: new Date() })
+        .where("id", "=", "test.lane_maint")
+        .execute();
+      expect(await sched.tick()).toContain("test.lane_maint");
+      await sched.idle();
+      expect(calls.get("test.lane_maint")).toBe(1);
+      expect(lanes.active("maintenance")).toBe(0);
+
+      // 到期未跑的:空闲后执行
+      await h.db
+        .updateTable("scheduled_job")
+        .set({ next_run_at: new Date(Date.now() - 1000) })
+        .where("id", "=", "test.lane_maint")
+        .execute();
+      expect(await sched.tick()).not.toContain("test.lane_maint");
+      busy = false;
+      expect(await sched.tick()).toContain("test.lane_maint");
+      await sched.idle();
+      expect(calls.get("test.lane_maint")).toBe(2);
+    } finally {
+      setLaneController(ctx, null);
+      reg.unregister("test.lane_maint");
+      reg.unregister("test.lane_light");
+    }
+  });
+
   it("登记默认值;用户改过的开关与调度不被再次登记覆盖;reset 恢复默认", async () => {
     registerScheduledJob(ctx, counting("test.keep"));
     await sched.tick();
diff --git a/apps/server/test/system/system.test.ts b/apps/server/test/system/system.test.ts
index f2c45d2e..31751f4d 100644
--- a/apps/server/test/system/system.test.ts
+++ b/apps/server/test/system/system.test.ts
@@ -51,6 +51,7 @@ describe("纯函数", () => {
         eventLoopLagMs: 0,
         eventLoopLagMaxMs: o.lag ?? 0,
         concurrency: {},
+        lanes: [] as never[],
         runningRuns: null,
         sseConnections: null,
         extra: {},
@@ -79,6 +80,7 @@ describe("纯函数", () => {
       eventLoopLagMs: 0,
       eventLoopLagMaxMs: 0,
       concurrency: {},
+      lanes: [] as never[],
       runningRuns: null,
       sseConnections: null,
       extra: {},
diff --git a/apps/server/test/worker-lanes.test.ts b/apps/server/test/worker-lanes.test.ts
new file mode 100644
index 00000000..6ac69d49
--- /dev/null
+++ b/apps/server/test/worker-lanes.test.ts
@@ -0,0 +1,319 @@
+import { readFileSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+import { type DB, migrateToLatest } from "@umt/db";
+import { defineTask, type TaskLane } from "@umt/tasks";
+import { Kysely, PostgresDialect } from "kysely";
+import pg from "pg";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { z } from "zod";
+import { ConfigSchema } from "../src/config.ts";
+import { type Bootstrapped, bootstrap, buildTaskRegistry } from "../src/context.ts";
+import { EventBus } from "../src/http/sse.ts";
+import { SETTINGS_CHANGED_EVENT } from "../src/modules/auth-settings/provided.ts";
+import { SERVER_MODULES } from "../src/modules/index.ts";
+import type { ServerContext, ServerModule } from "../src/modules/types.ts";
+import { type RunningWorker, startWorker } from "../src/worker/index.ts";
+import { groupTasksByLane, LaneController, laneOverridesFromEnv } from "../src/worker/lanes.ts";
+import { withSchemaLock } from "./auth-settings/harness.ts";
+import { memoryLogger } from "./helpers.ts";
+
+// WK-1 IO Worker 分道:归属表完整性(注册表 ↔ runbook)、控制器规则(并发热更新、让路、env 映射)、
+// 真实 PG 上三条道互不阻塞与维护道让路。
+
+const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
+const fakeCtx = { role: "all" } as unknown as ServerContext;
+
+/** docs/runbook/dev.md「任务归属表」:| `name` | lane | 说明 | */
+function runbookLaneTable(): Map<string, string> {
+  const md = readFileSync(fileURLToPath(new URL("../../../docs/runbook/dev.md", import.meta.url)), "utf8");
+  const section = md.slice(md.indexOf("#### 任务归属表"));
+  const end = section.indexOf("\n\n", section.indexOf("| 任务 |"));
+  const rows = section.slice(0, end).split("\n");
+  const out = new Map<string, string>();
+  for (const r of rows) {
+    const m = /^\| `([A-Za-z0-9_.-]+)` \| (\w+) \|/.exec(r);
+    if (m) out.set(m[1]!, m[2]!);
+  }
+  return out;
+}
+
+describe("任务归属表", () => {
+  const reg = buildTaskRegistry(SERVER_MODULES, fakeCtx);
+
+  it("注册表里每个任务都显式声明了 lane", () => {
+    const { missing } = groupTasksByLane(reg.all());
+    expect(missing, "这些任务没在 defineTask 里声明 lane").toEqual([]);
+  });
+
+  it("runbook 的归属表与注册表逐项一致(新增任务要同时写进 docs/runbook/dev.md)", () => {
+    const table = runbookLaneTable();
+    const actual = new Map(reg.all().map((d) => [d.name, d.lane as string]));
+    expect(Object.fromEntries(table)).toEqual(Object.fromEntries(actual));
+  });
+
+  it("没声明 lane:runAgent/resolveDecision 兜底归 agent,其余归网络道并列入 missing", () => {
+    const g = groupTasksByLane([{ name: "x" }, { name: "runAgent" }, { name: "f", lane: "file" }]);
+    expect(g.byLane.network).toEqual(["x"]);
+    expect(g.byLane.agent).toEqual(["runAgent"]);
+    expect(g.missing).toEqual(["runAgent", "x"]);
+  });
+});
+
+describe("LaneController 规则", () => {
+  const { logger } = memoryLogger();
+  const lanes = { file: ["f"], network: ["n"], maintenance: ["m"] };
+
+  it("旧 WORKER_CONCURRENCY 映射到网络道;新变量优先", () => {
+    expect(laneOverridesFromEnv({ WORKER_CONCURRENCY: 6 })).toEqual({ network: 6 });
+    expect(
+      laneOverridesFromEnv({ WORKER_CONCURRENCY: 6, WORKER_NETWORK_CONCURRENCY: 3, WORKER_FILE_CONCURRENCY: 1 }),
+    ).toEqual({ network: 3, file: 1 });
+    expect(laneOverridesFromEnv({})).toEqual({});
+  });
+
+  it("并发读设置、热更新;环境变量覆盖设置;槽位按上限开", async () => {
+    let settings = { file: 2, network: 4, maintenance: 1 };
+    const c = new LaneController({
+      engine: {},
+      logger,
+      lanes,
+      readSettings: async () => settings,
+      overrides: { maintenance: 20 },
+    });
+    await c.refresh();
+    expect([c.limit("file"), c.limit("network"), c.limit("maintenance")]).toEqual([2, 4, 20]);
+    expect([c.slots("file"), c.slots("maintenance")]).toEqual([16, 20]);
+    expect(c.source("maintenance")).toBe("env");
+    settings = { file: 5, network: 1, maintenance: 3 };
+    await c.refresh();
+    expect([c.limit("file"), c.limit("network"), c.limit("maintenance")]).toEqual([5, 1, 20]);
+  });
+
+  it("维护道:其它道有活让路,空闲放行;连续让路超时放行一个;立即执行不让路;内联占并发", async () => {
+    let busy = { queued: 1, running: 0 };
+    let now = 0;
+    const c = new LaneController({
+      engine: { queueCounts: async (groups) => groups.map(() => busy) },
+      logger,
+      lanes,
+      readSettings: async () => ({ file: 2, network: 4, maintenance: 1 }),
+      busyCacheMs: 0,
+      maxYieldMs: 1000,
+      now: () => now,
+    });
+    await c.refresh();
+    expect(await c.enterMaintenance()).toBeNull();
+    expect(c.yielding()).toBe(true);
+    now = 500;
+    expect(await c.maintenanceMayStart()).toBe(false);
+    now = 1000;
+    expect(await c.maintenanceMayStart()).toBe(true); // 让路满 1s 放行一个
+    now = 1200;
+    expect(await c.maintenanceMayStart()).toBe(false); // 重新计时
+    const forced = await c.enterMaintenance(true);
+    expect(forced).not.toBeNull();
+    expect(c.active("maintenance")).toBe(1);
+    expect(c.maxActive("maintenance")).toBe(0); // 引擎维护 worker 让出被内联占用的并发
+    forced?.();
+    busy = { queued: 0, running: 0 };
+    const r1 = await c.enterMaintenance();
+    expect(r1).not.toBeNull();
+    expect(c.yielding()).toBe(false);
+    expect(await c.enterMaintenance()).toBeNull(); // 并发 1 已被占
+    r1?.();
+    const snap = await c.snapshot();
+    expect(snap.map((s) => [s.lane, s.concurrency, s.running, s.queued])).toEqual([
+      ["file", 2, 0, 0],
+      ["network", 4, 0, 0],
+      ["maintenance", 1, 0, 0],
+    ]);
+  });
+});
+
+// ── 真实 PG ──
+
+const url = process.env.DATABASE_URL ?? "postgres://umt:umt@localhost:55433/umt";
+const uniq = `${process.pid.toString(36)}${Math.random().toString(36).slice(2, 7)}`;
+const biz = `server_lanes_${uniq}`;
+const ow = `${biz}_ow`;
+
+async function pgAvailable() {
+  const c = new pg.Client({ connectionString: url, connectionTimeoutMillis: 2000 });
+  try {
+    await c.connect();
+    await c.end();
+    return true;
+  } catch {
+    return false;
+  }
+}
+const hasPg = process.env.UMT_SKIP_DB_TESTS !== "1" && (await pgAvailable());
+
+async function waitUntil(fn: () => boolean | Promise<boolean>, ms = 15_000) {
+  const until = Date.now() + ms;
+  while (Date.now() < until) {
+    if (await fn()) return;
+    await sleep(50);
+  }
+  throw new Error("等待超时");
+}
+
+describe.skipIf(!hasPg)("io worker 分道(真实 PG)", () => {
+  let db: Kysely<DB>;
+  let booted: Bootstrapped;
+  let worker: RunningWorker;
+  const bus = new EventBus(50);
+  const started: string[] = [];
+  const done: string[] = [];
+  const gates = new Map<string, () => void>();
+
+  const task = (name: string, lane: TaskLane, block: boolean) =>
+    defineTask({
+      name,
+      version: 1,
+      lane,
+      inputSchema: z.object({ id: z.string() }),
+      retryPolicy: { maximumAttempts: 1 },
+      handler: async (t) => {
+        started.push(t.input.id);
+        if (block) await new Promise<void>((resolve) => gates.set(t.input.id, resolve));
+        done.push(t.input.id);
+        return null;
+      },
+    });
+  const probe: ServerModule = {
+    name: "probe",
+    registerRoutes() {},
+    registerTasks(registry) {
+      registry.register(
+        task("laneFile", "file", true),
+        task("laneNet", "network", false),
+        task("laneNetBlock", "network", true),
+        task("laneMaint", "maintenance", false),
+      );
+    },
+  };
+  const enqueue = (name: string, id: string) =>
+    booted.ctx.engine.enqueue(name, { id }, { idempotencyKey: `${name}:${id}`, traceId: id });
+
+  beforeAll(async () => {
+    const admin = new pg.Client({ connectionString: url });
+    await admin.connect();
+    await admin.query(`create schema ${biz}`);
+    await admin.end();
+    db = new Kysely<DB>({
+      dialect: new PostgresDialect({
+        pool: new pg.Pool({ connectionString: url, max: 8, options: `-c search_path=${biz},public` }),
+      }),
+    });
+    await withSchemaLock(() => migrateToLatest(db, { migrationTableSchema: biz }));
+    await db
+      .insertInto("settings")
+      .values({
+        key: "automation",
+        value: JSON.stringify({
+          enabled: true,
+          readOnlyReconcile: false,
+          workerLanes: { file: 1, network: 2, maintenance: 1 },
+        }),
+      })
+      .execute();
+    const config = ConfigSchema.parse({ DATABASE_URL: url, SECRET_MASTER_KEY: "test-master-key-0123456789" });
+    const { logger } = memoryLogger();
+    booted = await bootstrap({
+      role: "worker",
+      config,
+      logger,
+      db,
+      events: bus,
+      modules: [probe],
+      engineSchema: ow,
+      engineMigrations: true,
+    });
+    worker = await startWorker({
+      logger,
+      heartbeatMs: 60_000,
+      runtime: {
+        ctx: booted.ctx,
+        modules: [probe],
+        setRelay: booted.outbox.setRelay,
+        outboxIntervalMs: 100,
+        group: "io",
+        tasks: booted.tasks.all(),
+        lanes: { refreshMs: 100, busyCacheMs: 0 },
+      },
+    });
+  }, 60_000);
+
+  afterAll(async () => {
+    for (const g of gates.values()) g();
+    await worker?.stop();
+    await booted?.engine.stop();
+    await db?.destroy();
+    await withSchemaLock(async () => {
+      const admin = new pg.Client({ connectionString: url });
+      await admin.connect();
+      await admin.query(`drop schema if exists ${biz} cascade`);
+      await admin.query(`drop schema if exists ${ow} cascade`);
+      await admin.end();
+    });
+  }, 60_000);
+
+  it("文件道占满(并发 1)时网络道照常领取;维护道在文件道有活时让路,空了再跑;改设置热生效", async () => {
+    const lanes = worker.lanes!;
+    expect(lanes.limit("file")).toBe(1);
+
+    await enqueue("laneFile", "f1");
+    await enqueue("laneFile", "f2");
+    await waitUntil(() => started.includes("f1") || started.includes("f2"));
+    const first = started.includes("f1") ? "f1" : "f2";
+    const second = first === "f1" ? "f2" : "f1";
+
+    // 文件道占满:网络任务仍被领取并完成;第二个文件任务排队
+    await enqueue("laneNet", "n1");
+    await waitUntil(() => done.includes("n1"));
+    expect(started).not.toContain(second);
+
+    // 维护道让路:文件道有运行中 + 排队
+    await enqueue("laneMaint", "m1");
+    await sleep(1500);
+    expect(started).not.toContain("m1");
+    expect(lanes.yielding()).toBe(true);
+    const snap = await lanes.snapshot();
+    expect(snap.find((s) => s.lane === "file")).toMatchObject({ concurrency: 1, running: 1, queued: 1 });
+    expect(snap.find((s) => s.lane === "maintenance")).toMatchObject({ queued: 1, yielding: true });
+
+    // 热生效:文件道并发改 2,排队的第二个文件任务被领取
+    await db
+      .updateTable("settings")
+      .set({
+        value: JSON.stringify({
+          enabled: true,
+          readOnlyReconcile: false,
+          workerLanes: { file: 2, network: 2, maintenance: 1 },
+        }),
+      })
+      .where("key", "=", "automation")
+      .execute();
+    await bus.publish(SETTINGS_CHANGED_EVENT, { keys: ["automation"] });
+    await waitUntil(() => started.includes(second));
+    expect(lanes.limit("file")).toBe(2);
+    expect(started).not.toContain("m1");
+
+    // 文件道清空 → 维护道领取
+    gates.get(first)?.();
+    gates.get(second)?.();
+    await waitUntil(() => done.includes("m1"));
+    expect(lanes.yielding()).toBe(false);
+  }, 60_000);
+
+  it("网络道占满时文件道照常领取", async () => {
+    await enqueue("laneNetBlock", "nb1");
+    await enqueue("laneNetBlock", "nb2");
+    await waitUntil(() => started.includes("nb1") && started.includes("nb2"));
+    await enqueue("laneFile", "f3");
+    await waitUntil(() => started.includes("f3"));
+    for (const id of ["nb1", "nb2", "f3"]) gates.get(id)?.();
+    await waitUntil(() => ["nb1", "nb2", "f3"].every((id) => done.includes(id)));
+  }, 60_000);
+});
diff --git a/apps/web/app/components/settings/automation.tsx b/apps/web/app/components/settings/automation.tsx
index bb26eb02..5be28d13 100644
--- a/apps/web/app/components/settings/automation.tsx
+++ b/apps/web/app/components/settings/automation.tsx
@@ -10,7 +10,7 @@ import { type FormValues, num } from "./resource-section.tsx";
 import { anchorId, SettingsBlock } from "./settings-block.tsx";
 import { type CardSpec, n, SettingsCard, SettingsGate } from "./settings-card.tsx";
 
-// 自动化分区:总开关、只读对账、缓存水位、每日下载预算、保留天数、存储预算与保留(CACHE-2)
+// 自动化分区:总开关、只读对账、缓存水位、每日下载预算、保留天数、后台任务并发(WK-1)、存储预算与保留(CACHE-2)
 
 const GiB = 1024 ** 3;
 
@@ -106,6 +106,33 @@ const dayField = (group: "retentionDays" | "cacheTtl", k: string, label: string,
 const pick = (v: FormValues, prefix: string, keys: readonly string[]) =>
   Object.fromEntries(keys.map((k) => [k, num(v[`${prefix}.${k}`])]));
 
+// WK-1:io worker 三条道的并发(settings.automation.workerLanes,docs/runbook/dev.md「IO Worker 分道」)
+const LANE_KEYS = ["file", "network", "maintenance"] as const;
+const WORKER_LANES_CARD: CardSpec = {
+  anchor: "worker-lanes",
+  title: "后台任务并发",
+  description:
+    "后台任务分三条道各自排队，互不占用。保存后几秒内生效，不用重启；服务器上设置了对应的环境变量时以环境变量为准。",
+  fields: [
+    { name: "automation.workerLanes.file", label: "文件道", kind: "number", hint: "整理、硬链接、移动、校验；1–16" },
+    {
+      name: "automation.workerLanes.network",
+      label: "网络道",
+      kind: "number",
+      hint: "站点、RSS、TMDB、下载器、搜索；单站限流另计；1–16",
+    },
+    {
+      name: "automation.workerLanes.maintenance",
+      label: "维护道",
+      kind: "number",
+      hint: "清理、回填、对账；前两道有活时让路；1–16",
+    },
+  ],
+  fromSettings: (s) =>
+    Object.fromEntries(LANE_KEYS.map((k) => [`automation.workerLanes.${k}`, n(s.automation.workerLanes[k])])),
+  toPatch: (v) => ({ automation: { workerLanes: pick(v, "automation.workerLanes", LANE_KEYS) } }),
+};
+
 const BUDGET_KEYS = ["pgDataGb", "pgDataWarnGb", "imagesGb", "backupsGb", "sessionsEvidenceGb"] as const;
 const STORAGE_BUDGET_CARD: CardSpec = {
   anchor: "storage",
@@ -212,7 +239,7 @@ function ToggleRow({
   );
 }
 
-/** 高级:只读对账、存储预算、流水保留、缓存有效期;默认收起,从 #automation/advanced 进来时展开 */
+/** 高级:只读对账、后台任务并发、存储预算、流水保留、缓存有效期;默认收起,从 #automation/advanced 进来时展开 */
 function Advanced({ children }: { children: ReactNode }) {
   const location = useLocation();
   const [open, setOpen] = useState(() => location.hash.endsWith("/advanced"));
@@ -232,7 +259,9 @@ function Advanced({ children }: { children: ReactNode }) {
           className={cn("size-4 text-muted-foreground transition-transform duration-base", open && "rotate-180")}
         />
       </button>
-      <p className="mt-1 text-caption text-muted-foreground">只读对账、存储预算、数据保留和缓存有效期。一般不用改。</p>
+      <p className="mt-1 text-caption text-muted-foreground">
+        只读对账、后台任务并发、存储预算、数据保留和缓存有效期。一般不用改。
+      </p>
       {open ? <div className="mt-6 grid grid-cols-1 gap-8">{children}</div> : null}
     </section>
   );
@@ -309,6 +338,7 @@ export function AutomationSection() {
             <SettingsCard section="automation" spec={RETENTION_CARD} settings={s} />
             <Advanced>
               <ReadOnlySwitch />
+              <SettingsCard section="automation" spec={WORKER_LANES_CARD} settings={s} />
               <SettingsCard section="automation" spec={STORAGE_BUDGET_CARD} settings={s} />
               <SettingsCard section="automation" spec={STORAGE_RETENTION_CARD} settings={s} />
               <SettingsCard section="automation" spec={STORAGE_TTL_CARD} settings={s} />
diff --git a/apps/web/app/routes/console/status.tsx b/apps/web/app/routes/console/status.tsx
index d7b8f4ae..f6db602d 100644
--- a/apps/web/app/routes/console/status.tsx
+++ b/apps/web/app/routes/console/status.tsx
@@ -1,5 +1,5 @@
 import { useQueryClient } from "@tanstack/react-query";
-import type { HealthLevel, SystemHealth, SystemProcess } from "@umt/contracts";
+import type { HealthLevel, SystemHealth, SystemProcess, SystemWorkerLane } from "@umt/contracts";
 import {
   AlertTriangle,
   ChevronDown,
@@ -226,6 +226,8 @@ function Processes({ d }: { d: SystemHealth }) {
           cell: (p) =>
             p.concurrency.io === undefined && p.concurrency.agent === undefined ? (
               <span className="text-muted-foreground">—</span>
+            ) : p.lanes.length ? (
+              <LaneCell p={p} />
             ) : (
               <span className="whitespace-nowrap tabular-nums">
                 io {p.concurrency.io ?? 0} · 智能体 {p.concurrency.agent ?? 0}
@@ -261,6 +263,30 @@ function Processes({ d }: { d: SystemHealth }) {
   );
 }
 
+const LANE_LABEL: Record<SystemWorkerLane["lane"], string> = { file: "文件", network: "网络", maintenance: "维护" };
+
+/** io worker 分道(WK-1):每道一行「运行中/并发 · 排队」,维护道让路时标出来 */
+function LaneCell({ p }: { p: SystemProcess }) {
+  return (
+    <div className="grid justify-items-end gap-0.5 text-meta">
+      {p.lanes.map((l) => (
+        <span
+          key={l.lane}
+          className="whitespace-nowrap tabular-nums"
+          title={l.source === "env" ? "并发由环境变量指定，设置页的值不生效" : undefined}
+        >
+          {LANE_LABEL[l.lane]} {l.running}/{l.concurrency}
+          {l.queued ? ` · 排队 ${l.queued}` : ""}
+          {l.yielding ? " · 让路中" : ""}
+        </span>
+      ))}
+      {p.concurrency.agent !== undefined ? (
+        <span className="whitespace-nowrap tabular-nums text-muted-foreground">智能体 {p.concurrency.agent}</span>
+      ) : null}
+    </div>
+  );
+}
+
 /** 已下线实例:已退出、容器已重建或被同容器新进程替换的旧心跳行,默认折叠,不参与告警 */
 function RetiredProcesses({ rows }: { rows: SystemProcess[] }) {
   const [open, setOpen] = useState(false);
diff --git a/docker-compose.yml b/docker-compose.yml
index 588e1bca..6829f3a2 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -177,8 +177,13 @@ services:
       PUID: ${PUID:-10001}
       PGID: ${PGID:-10001}
       BACKUP_KEEP: ${BACKUP_KEEP:-14}
-      # worker 并发:io 组(下载提交/入库复制/清理/对账/抓取等)与 agent 组(runAgent/resolveDecision)分池
-      WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-4}
+      # worker 并发:io 组(文件 / 网络 / 维护三条道)与 agent 组(runAgent/resolveDecision)分池
+      # io worker 三条道的并发(WK-1):不设 = 用 设置 → 自动化 → 高级 里的值(缺省 文件 2 / 网络 4 / 维护 1);
+      # 设了就以环境变量为准。旧 WORKER_CONCURRENCY 映射到网络道。
+      WORKER_CONCURRENCY: ${WORKER_CONCURRENCY:-}
+      WORKER_FILE_CONCURRENCY: ${WORKER_FILE_CONCURRENCY:-}
+      WORKER_NETWORK_CONCURRENCY: ${WORKER_NETWORK_CONCURRENCY:-}
+      WORKER_MAINTENANCE_CONCURRENCY: ${WORKER_MAINTENANCE_CONCURRENCY:-}
       WORKER_AGENT_CONCURRENCY: ${WORKER_AGENT_CONCURRENCY:-2}
       # 手动按路径整理时递归遍历的文件数上限
       ORGANIZE_WALK_MAX_FILES: ${ORGANIZE_WALK_MAX_FILES:-5000}
diff --git a/docker/compose.env.example b/docker/compose.env.example
index 8e3ea749..d71061fd 100644
--- a/docker/compose.env.example
+++ b/docker/compose.env.example
@@ -29,8 +29,12 @@ LOG_LEVEL=info
 # APP_VERSION=v0.1.0
 # 国内网络构建换源(可选):APT_MIRROR=mirrors.ustc.edu.cn
 
-# worker 并发(可选):io 组(下载提交/入库复制/缓存清理/资产对账/抓取)与 agent 组(Agent 会话)分池
-# WORKER_CONCURRENCY=4
+# worker 并发(可选):io 组分文件 / 网络 / 维护三条道,agent 组(Agent 会话)单独一池。
+# 三条道缺省用 设置 → 自动化 → 高级 里的值(可热改);设了下面的变量就以变量为准。
+# 旧 WORKER_CONCURRENCY 映射到网络道。
+# WORKER_FILE_CONCURRENCY=2
+# WORKER_NETWORK_CONCURRENCY=4
+# WORKER_MAINTENANCE_CONCURRENCY=1
 # WORKER_AGENT_CONCURRENCY=2
 # 手动按路径整理时递归遍历的文件数上限(dryRun 超出返回 truncated,正式执行拒绝)
 # ORGANIZE_WALK_MAX_FILES=5000
diff --git a/docs/runbook/deploy.md b/docs/runbook/deploy.md
index bbae529b..9a95dbdf 100644
--- a/docs/runbook/deploy.md
+++ b/docs/runbook/deploy.md
@@ -124,7 +124,8 @@ docker compose logs app | grep 'setup token'
 | `PUID` / `PGID` | `10001` | 运行用户 |
 | `DATA_DIR` | `/data` | 应用数据目录(开发机缺省 `./data`) |
 | `AGENT_SESSION_DIR` | `DATA_DIR/agent-sessions` | Agent 会话目录 |
-| `WORKER_CONCURRENCY` | `4` | io 组 worker 并发槽 |
+| `WORKER_FILE_CONCURRENCY` / `WORKER_NETWORK_CONCURRENCY` / `WORKER_MAINTENANCE_CONCURRENCY` | 空 | io worker 文件 / 网络 / 维护三条道的并发;空 = 用 设置 → 自动化 → 高级 →「后台任务并发」(缺省 2 / 4 / 1,热生效),设了以变量为准。见 dev.md「IO Worker 分道」 |
+| `WORKER_CONCURRENCY` | 空 | 旧变量(原 io 组总并发),现映射到网络道(`WORKER_NETWORK_CONCURRENCY` 未设时) |
 | `WORKER_AGENT_CONCURRENCY` | `2` | agent 组 worker 并发槽(每个 Agent 会话占一槽直到结束) |
 | `ORGANIZE_WALK_MAX_FILES` | `5000` | 手动按路径整理时递归遍历的文件数上限;dryRun 超出返回 `truncated: true`,正式执行拒绝 |
 | `BACKUP_KEEP` | `14` | 备份保留份数,0 不轮转 |
diff --git a/docs/runbook/dev.md b/docs/runbook/dev.md
index b58f45a2..91ed5456 100644
--- a/docs/runbook/dev.md
+++ b/docs/runbook/dev.md
@@ -52,12 +52,68 @@ curl -H "Authorization: Bearer umt-dev-admin-key" "localhost:4380/api/v1/events?
 | 角色 | 引擎 worker | OutboxRelay / 模块后台循环 |
 |---|---|---|
 | `api` | 不起(只入队/信号/取消) | 不跑 |
-| `worker`(io 组) | 除 `runAgent`/`resolveDecision` 外的全部任务,并发 `WORKER_CONCURRENCY`(缺省 4) | 跑 |
-| `worker:agent` | 只跑 `runAgent`/`resolveDecision`,并发 `WORKER_AGENT_CONCURRENCY`(缺省 2) | 不跑 |
-| `all` | io 组 + agent 组两个池(开发缺省) | 跑 |
+| `worker`(io 组) | 除 agent 道外的全部任务,分文件 / 网络 / 维护三条道(见下) | 跑 |
+| `worker:agent` | 只跑 agent 道(`runAgent`/`resolveDecision`),并发 `WORKER_AGENT_CONCURRENCY`(缺省 2) | 不跑 |
+| `all` | io 三条道 + agent 共四个池(开发缺省) | 跑 |
 
 分组 worker 只认领自己的任务(`@umt/tasks` 按任务名过滤认领),Agent 长会话不占文件复制的槽,反之亦然。
-worker 里有事件循环延迟探针:偏差超过 5s 打 `worker 事件循环延迟超过阈值` warn,次数随心跳事件(`loopLagWarnings`)发出。
+worker 里有事件循环延迟探针:偏差超过 5s 打 `worker 事件循环延迟超过阈值` warn,次数随心跳事件(`loopLagWarnings`)发出;
+每 15s 的进程心跳另记延迟与峰值(`process_heartbeat` / `process_heartbeat_sample`),运行状态页可看历史。
+
+### IO Worker 分道(WK-1)
+
+io worker 进程内按任务声明的 `lane` 起三个引擎 worker,各自独立并发、各自排队,一条道占满不影响另一条道领取:
+
+| 道 | 缺省并发 | 放什么 |
+|---|---|---|
+| 文件 `file` | 2 | 整理(复制/硬链接/移动)、ffprobe、校验、接管整理、下载盘阈值清理 |
+| 网络 `network` | 4 | 站点快照、RSS、TMDB、元数据刷新、搜索、下载器提交、识别(单站限流照旧另计,见 sites.md) |
+| 维护 `maintenance` | 1 | 回收站清理、资产对账、调度器里内联的清理 / 回填 / 保留期裁剪;**空闲时跑** |
+
+- **声明**:任务在 `defineTask({ lane })` 里显式声明(`@umt/tasks` 的 `TASK_LANES`)。没声明的任务运行时归网络道并在启动日志里
+  warn「这些任务没有声明 lane」;ci 里 `apps/server/test/worker-lanes.test.ts` 会失败(注册表与下表逐项核对,新增任务必须同时改这里)。
+  选"ci 报错 + 运行时兜底网络道"而不是"启动即失败":漏声明不该让生产进程起不来,但不能悄悄进 main。
+- **维护道让路**:维护道每次领取前查全库文件道 + 网络道有没有已到点排队或正在运行的 run(1s 缓存),有就不领;已经在跑的维护任务
+  不打断。连续让路超过 10 分钟放行一个(日志「维护道让路超时」),防止网络道长期有活时维护永远跑不上。实现是给维护道 worker 的
+  认领加闸门(`StartWorkerOptions.claimGate`),不改 OpenWorkflow;取舍:按"全库有无活"判断,多个 io 实例时别的实例忙也会让路,
+  判断有 1s 滞后,够用且简单。
+- **调度器内联的维护任务**:`ScheduledJobDef.lane = "maintenance"` 的定时任务(下表)在调度器里直接执行,与维护道共用并发与让路规则:
+  进不了维护道就保持到期、下一轮再试;在计划页点「立即执行」不让路。其余定时任务只入队或很快结束,不占道。
+- **并发配置**:设置 → 自动化 → 高级 →「后台任务并发」(`settings.automation.workerLanes`),保存后约 5 秒内热生效(每道开 16 个槽,
+  按设置值收)。环境变量 `WORKER_FILE_CONCURRENCY` / `WORKER_NETWORK_CONCURRENCY` / `WORKER_MAINTENANCE_CONCURRENCY` 设了就以变量为准
+  (运行状态页悬停提示"由环境变量指定")。旧变量 `WORKER_CONCURRENCY` 映射到**网络道**(`WORKER_NETWORK_CONCURRENCY` 未设时):它原来
+  是 io 组总并发,缺省 4,与网络道缺省相同;文件 / 维护道按设置页走。compose 不再给 `WORKER_CONCURRENCY` 缺省值。
+- **观测**:运行状态页「进程」表的并发列,io 进程按道显示「运行中/并发 · 排队 N」,维护道让路时标「让路中」;数据在心跳行
+  `concurrency.{file,network,maintenance}` 与 `extra.lanes`(排队是全库数,运行中是本进程数)。
+- **拆进程的触发条件**(未实施):整理期间心跳事件循环延迟峰值持续 > 200ms,或整理需要贴近存储部署到另一台机器。
+
+#### 任务归属表
+
+ci 逐行核对这张表与任务注册表(`apps/server/test/worker-lanes.test.ts`),改任务的 lane 或新增任务时同步改。
+
+| 任务 | 道 | 说明 |
+|---|---|---|
+| `verifyAndImport` | file | 下载完成后的校验、ffprobe、复制/硬链接/移动入库(含接管整理) |
+| `cleanupDownloadCache` | file | 下载盘到暂停线时的清理;要尽快腾空间,不放维护道 |
+| `ingestFeed` | network | RSS 抓取 |
+| `resolveObservation` | network | 标题识别 + TMDB |
+| `enrichPoolEntry` | network | 未识别资源的元数据补全 |
+| `refreshMetadata` | network | TMDB 元数据刷新 |
+| `searchSubscription` | network | 订阅搜索 |
+| `snapshotSite` | network | 站点快照 |
+| `refreshSiteAccount` | network | 站点登录 / 用户数据刷新 |
+| `evaluateAcquisition` | network | 获取策略评估(查库为主,随上游链路放网络道) |
+| `submitDownload` | network | 提交下载器 |
+| `identifyRemote` | network | 文件管理「识别」的 TMDB 查询 |
+| `refreshPiTokens` | network | 模型服务 OAuth 令牌刷新 |
+| `reconcileLibrary` | maintenance | 资产对账(扫库) |
+| `purgeTrash` | maintenance | 回收站到期清除 |
+| `runAgent` | agent | Agent 会话(worker:agent) |
+| `resolveDecision` | agent | 决策处理(worker:agent) |
+
+调度器内联、走维护道的定时任务:`cache.sweep`、`cache.images_clean`、`cache.images_lru`、`auth.cleanup`、`system.heartbeat_cleanup`、
+各保留期裁剪(`system/retention.ts` 的 `RETENTION_JOBS`)、`metadata.backfill_languages`(元数据语言回刷)。
+只入队的定时任务(如 `trash.purge`、`cache.threshold`、`library.reconcile`、`pi.token_refresh`)按所入队任务的道执行。
 
 启动时 `bootstrap()` 装配 `apps/server/src/modules/` 下的四个业务模块(见 apps/server/README.md),并连接任务引擎。数据库不可达时进程启动失败并退出。
 
diff --git a/packages/contracts/src/ops.ts b/packages/contracts/src/ops.ts
index 071811fd..3909a62e 100644
--- a/packages/contracts/src/ops.ts
+++ b/packages/contracts/src/ops.ts
@@ -375,6 +375,24 @@ export type ProcessKind = z.infer<typeof ProcessKind>;
 export const HealthLevel = z.enum(["ok", "warn", "critical", "unknown"]);
 export type HealthLevel = z.infer<typeof HealthLevel>;
 
+export const WorkerLane = z.enum(["file", "network", "maintenance"]);
+export type WorkerLane = z.infer<typeof WorkerLane>;
+
+export const SystemWorkerLane = z.object({
+  lane: WorkerLane,
+  /** 当前生效的并发上限 */
+  concurrency: NonNegInt,
+  /** env = 环境变量覆盖(设置页改了不生效);settings = 设置页的值 */
+  source: z.enum(["env", "settings"]),
+  /** 本进程该道正在执行的数 */
+  running: NonNegInt,
+  /** 全库该道已到点等待领取的数;统计失败为 null */
+  queued: NonNegInt.nullable(),
+  /** 维护道:正在给文件道 / 网络道让路 */
+  yielding: z.boolean().optional(),
+});
+export type SystemWorkerLane = z.infer<typeof SystemWorkerLane>;
+
 export const SystemProcess = z.object({
   /** <kind>:<host>:<pid>;同一容器重启覆盖同一行 */
   instanceId: z.string(),
@@ -394,8 +412,16 @@ export const SystemProcess = z.object({
   eventLoopLagMs: NonNegInt,
   /** 最近一个心跳周期(15s)内的最大延迟 */
   eventLoopLagMaxMs: NonNegInt,
-  /** 引擎 worker 并发;api 进程为 {} */
-  concurrency: z.object({ io: NonNegInt.optional(), agent: NonNegInt.optional() }),
+  /** 引擎 worker 并发;api 进程为 {}。io = 三条道合计(WK-1 前是 io 组总并发) */
+  concurrency: z.object({
+    io: NonNegInt.optional(),
+    agent: NonNegInt.optional(),
+    file: NonNegInt.optional(),
+    network: NonNegInt.optional(),
+    maintenance: NonNegInt.optional(),
+  }),
+  /** io worker 各道(WK-1):并发、本进程运行中、全库排队;非 io 进程或旧心跳为 [] */
+  lanes: z.array(SystemWorkerLane).default([]),
   /** 本组任务正在执行的 run 数(同组多实例时是组合计);api 进程为 null */
   runningRuns: NonNegInt.nullable(),
   /** 本进程 SSE 长连接数;不托管 HTTP 的进程为 null */
diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts
index db47993f..628ffe5d 100644
--- a/packages/contracts/src/settings.ts
+++ b/packages/contracts/src/settings.ts
@@ -8,11 +8,31 @@ import { SiteRateLimit } from "./sources.ts";
 
 const Percent = z.number().min(1).max(100);
 
+/** 并发槽上限(与 io worker 各道开的槽位数一致;改设置只在这个范围内热生效) */
+export const WORKER_LANE_MAX = 16;
+/**
+ * io worker 三条道的并发(WK-1,docs/runbook/dev.md「IO Worker 分道」):改后约 5 秒内热生效,不用重启;
+ * 设置了环境变量 WORKER_FILE_CONCURRENCY / WORKER_NETWORK_CONCURRENCY(旧 WORKER_CONCURRENCY)/
+ * WORKER_MAINTENANCE_CONCURRENCY 时以环境变量为准。
+ */
+export const WorkerLaneSettings = z.object({
+  /** 文件道:整理、硬链接/移动、ffprobe、校验 */
+  file: z.number().int().min(1).max(WORKER_LANE_MAX),
+  /** 网络道:站点、RSS、TMDB、下载器、搜索(单站限流另计) */
+  network: z.number().int().min(1).max(WORKER_LANE_MAX),
+  /** 维护道:清理、回填、对账;文件道 / 网络道有活时让路 */
+  maintenance: z.number().int().min(1).max(WORKER_LANE_MAX),
+});
+export type WorkerLaneSettings = z.infer<typeof WorkerLaneSettings>;
+export const DEFAULT_WORKER_LANES: WorkerLaneSettings = { file: 2, network: 4, maintenance: 1 };
+
 export const AutomationSettings = z.object({
   /** 自动化总开关:关闭后所有自动收录/下载/整理停止,只读查询与手动操作可用 */
   enabled: z.boolean(),
   /** 恢复后的只读对账模式(REQ §8.5):开启时禁止自动写入 */
   readOnlyReconcile: z.boolean(),
+  /** io worker 分道并发;旧库缺省时补 DEFAULT_WORKER_LANES */
+  workerLanes: WorkerLaneSettings.default(DEFAULT_WORKER_LANES),
 });
 export type AutomationSettings = z.infer<typeof AutomationSettings>;
 
@@ -564,7 +584,7 @@ export type SettingsKey = z.infer<typeof SettingsKey>;
 
 /** 缺省设置(settings 表缺 key 或缺字段时使用) */
 export const DEFAULT_SETTINGS: GlobalSettings = {
-  automation: { enabled: true, readOnlyReconcile: false },
+  automation: { enabled: true, readOnlyReconcile: false, workerLanes: DEFAULT_WORKER_LANES },
   cache: {
     hardThresholdPercent: 90,
     resumePercent: 85,
@@ -646,7 +666,10 @@ export const DEFAULT_SETTINGS: GlobalSettings = {
 
 /** PATCH /settings:按分组部分更新;服务端与现值合并后用 GlobalSettings 整体再校验 */
 export const GlobalSettingsUpdate = z.object({
-  automation: AutomationSettings.partial().optional(),
+  automation: AutomationSettings.omit({ workerLanes: true })
+    .partial()
+    .extend({ workerLanes: WorkerLaneSettings.partial().optional() })
+    .optional(),
   cache: CacheShape.partial().optional(),
   organize: OrganizeSettings.partial().optional(),
   retention: RetentionSettings.partial().optional(),
@@ -709,7 +732,11 @@ export function mergeSettings(current: GlobalSettings, patch: GlobalSettingsUpda
   const { webSearch: wsPatch, ...agent } = patch.agent ?? {};
   const { apiKey: _wsApiKey, ...webSearch } = wsPatch ?? {};
   return {
-    automation: { ...current.automation, ...patch.automation },
+    automation: {
+      ...current.automation,
+      ...patch.automation,
+      workerLanes: { ...DEFAULT_WORKER_LANES, ...current.automation.workerLanes, ...patch.automation?.workerLanes },
+    },
     cache: { ...current.cache, ...patch.cache },
     organize: { ...current.organize, ...patch.organize },
     retention: { ...current.retention, ...patch.retention },
@@ -825,6 +852,7 @@ export const SETTINGS_SECTIONS: readonly SettingsSectionDef[] = [
     fields: [
       "automation.enabled",
       "automation.readOnlyReconcile",
+      "automation.workerLanes",
       "cache.hardThresholdPercent",
       "cache.resumePercent",
       "cache.checkIntervalSec",
diff --git a/packages/contracts/test/schemas.test.ts b/packages/contracts/test/schemas.test.ts
index 4cdc4a79..97ac2109 100644
--- a/packages/contracts/test/schemas.test.ts
+++ b/packages/contracts/test/schemas.test.ts
@@ -176,7 +176,15 @@ describe("关键合同行为", () => {
     const patch = C.GlobalSettingsUpdate.parse({ automation: { enabled: false }, pi: { contextBudget: 64_000 } });
     expect(patch).toEqual({ automation: { enabled: false }, pi: { contextBudget: 64_000 } });
     const merged = C.mergeSettings(C.DEFAULT_SETTINGS, patch);
-    expect(merged.automation).toEqual({ enabled: false, readOnlyReconcile: false });
+    expect(merged.automation).toEqual({
+      enabled: false,
+      readOnlyReconcile: false,
+      workerLanes: C.DEFAULT_WORKER_LANES,
+    });
+    // 分道并发部分更新:只改一道,其余保留
+    const lanes = C.mergeSettings(merged, C.GlobalSettingsUpdate.parse({ automation: { workerLanes: { file: 3 } } }));
+    expect(lanes.automation.workerLanes).toEqual({ file: 3, network: 4, maintenance: 1 });
+    expect(C.GlobalSettings.safeParse(lanes).success).toBe(true);
     expect(merged.pi).toEqual({ ...C.DEFAULT_SETTINGS.pi, contextBudget: 64_000 });
     expect(C.GlobalSettings.safeParse(merged).success).toBe(true);
   });
diff --git a/packages/tasks/src/definitions.ts b/packages/tasks/src/definitions.ts
index 6fc0b9db..1cc1e0a2 100644
--- a/packages/tasks/src/definitions.ts
+++ b/packages/tasks/src/definitions.ts
@@ -33,6 +33,7 @@ const DB_RETRY = { maximumAttempts: 3, initialInterval: "1s", backoffCoefficient
 export const ingestFeed = defineTask({
   name: "ingestFeed",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ feedSourceId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("ingestFeed"),
@@ -41,6 +42,7 @@ export const ingestFeed = defineTask({
 export const resolveObservation = defineTask({
   name: "resolveObservation",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ observationId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("resolveObservation"),
@@ -49,6 +51,7 @@ export const resolveObservation = defineTask({
 export const evaluateAcquisition = defineTask({
   name: "evaluateAcquisition",
   version: 1,
+  lane: "network",
   inputSchema: z.object({
     subscriptionId: id.optional(),
     acquisitionSlotId: id.optional(),
@@ -61,6 +64,7 @@ export const evaluateAcquisition = defineTask({
 export const submitDownload = defineTask({
   name: "submitDownload",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ acquisitionPlanId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("submitDownload"),
@@ -69,6 +73,7 @@ export const submitDownload = defineTask({
 export const verifyAndImport = defineTask({
   name: "verifyAndImport",
   version: 1,
+  lane: "file",
   inputSchema: z.object({ downloadTaskId: id }),
   retryPolicy: FILE_RETRY,
   handler: stub("verifyAndImport"),
@@ -77,6 +82,7 @@ export const verifyAndImport = defineTask({
 export const refreshMetadata = defineTask({
   name: "refreshMetadata",
   version: 1,
+  lane: "network",
   inputSchema: z.object({ mediaEntityId: id }),
   retryPolicy: NETWORK_RETRY,
   handler: stub("refreshMetadata"),
@@ -85,6 +91,7 @@ export const refreshMetadata = defineTask({
 export const reconcileLibrary = defineTask({
   name: "reconcileLibrary",
   version: 1,
+  lane: "maintenance",
   inputSchema: z.object({ libraryRootId: id.optional(), reconcileRunId: id.optional() }),
   retryPolicy: FILE_RETRY,
   handler: stub("reconcileLibrary"),
@@ -93,6 +100,7 @@ export const reconcileLibrary = defineTask({
 export const cleanupDownloadCache = defineTask({
   name: "cleanupDownloadCache",
   version: 1,
+  lane: "file",
   inputSchema: z.object({ downloaderInstanceId: id.optional() }),
   retryPolicy: FILE_RETRY,
   handler: stub("cleanupDownloadCache"),
@@ -101,6 +109,7 @@ export const cleanupDownloadCache = defineTask({
 export const resolveDecision = defineTask({
   name: "resolveDecision",
   version: 1,
+  lane: "agent",
   inputSchema: z.object({ decisionId: id }),
   retryPolicy: DB_RETRY,
   handler: stub("resolveDecision"),
diff --git a/packages/tasks/src/index.ts b/packages/tasks/src/index.ts
index fcc20859..b8951889 100644
--- a/packages/tasks/src/index.ts
+++ b/packages/tasks/src/index.ts
@@ -48,5 +48,13 @@ export {
   type OutboxRoute,
   type OutboxRouteFn,
 } from "./outbox-relay.ts";
-export { defineTask, type TaskDefinition, TaskRegistry } from "./registry.ts";
+export {
+  DEFAULT_TASK_LANE,
+  defineTask,
+  laneOf,
+  TASK_LANES,
+  type TaskDefinition,
+  type TaskLane,
+  TaskRegistry,
+} from "./registry.ts";
 export type * from "./types.ts";
diff --git a/packages/tasks/src/openworkflow-engine.ts b/packages/tasks/src/openworkflow-engine.ts
index 4cefff0c..9d26bfd6 100644
--- a/packages/tasks/src/openworkflow-engine.ts
+++ b/packages/tasks/src/openworkflow-engine.ts
@@ -9,6 +9,7 @@ import type {
   EnqueueOptions,
   JsonValue,
   ListFilter,
+  QueueCounts,
   StartWorkerOptions,
   TaskContext,
   TaskEngine,
@@ -16,6 +17,7 @@ import type {
   TaskRunView,
   TaskStepApi,
   WaitForFactOptions,
+  WorkerHandle,
 } from "./types.ts";
 
 // biome-ignore lint/suspicious/noExplicitAny: 本包只用原生 SQL 访问自有表
@@ -59,19 +61,32 @@ export async function migrateEngineSchema(databaseUrl: string, schema = ENGINE_S
   await backend.stop();
 }
 
+interface ClaimWrapOptions {
+  /** 只认领这些 "名称@引擎版本";null = 不按任务过滤 */
+  keys: string[] | null;
+  /** 本 worker 正在执行的 run 数 */
+  active: () => number;
+  maxActive?: () => number;
+  gate?: () => boolean | Promise<boolean>;
+}
+
 /**
- * 包一层只认领指定 (任务名, 版本) 的 backend:与 BackendPostgres.claimWorkflowRun 同一条 SQL(截止过期处理 +
- * SKIP LOCKED 认领),candidate 多一个 workflow_name/version 过滤。其余方法经原型链委托给原 backend。
+ * 包一层 backend 的认领:
+ * - keys:与 BackendPostgres.claimWorkflowRun 同一条 SQL(截止过期处理 + SKIP LOCKED 认领),candidate 多一个
+ *   workflow_name/version 过滤(OpenWorkflow 0.10.1 认领不看任务名,认领到未注册的 run 会以"未注册"失败并退避);
+ * - maxActive:执行中 + 认领中 ≥ 上限就不认领(同一轮 tick 的多个槽并发认领,用 reserved 计数挡住超额);
+ * - gate:返回 false 不认领。
+ * 其余方法经原型链委托给原 backend。
  */
-function claimFilteredBackend(backend: BackendPostgres, defs: readonly TaskDefinition[]): BackendPostgres {
+function claimWrappedBackend(backend: BackendPostgres, o: ClaimWrapOptions): BackendPostgres {
   // biome-ignore lint/suspicious/noExplicitAny: 访问 BackendPostgres 的私有字段(pg / namespaceId / workflowRunsTable)
   const b = backend as any;
-  const keys = [...new Set(defs.map((d) => `${d.name}@${engineVersion(d.version)}`))];
   const filtered = Object.create(backend) as BackendPostgres;
-  Object.defineProperty(filtered, "claimWorkflowRun", {
-    value: async (params: { workerId: string; leaseDurationMs: number }) => {
-      const table = b.workflowRunsTable();
-      const [claimed] = await b.pg`
+  let reserved = 0;
+  const claim = async (params: { workerId: string; leaseDurationMs: number }) => {
+    if (!o.keys) return backend.claimWorkflowRun(params);
+    const table = b.workflowRunsTable();
+    const [claimed] = await b.pg`
       WITH expired AS (
         UPDATE ${table}
         SET "status" = 'failed', "error" = ${b.pg.json({ message: "Workflow run deadline exceeded" })},
@@ -89,7 +104,7 @@ function claimFilteredBackend(backend: BackendPostgres, defs: readonly TaskDefin
           AND "status" IN ('pending', 'running', 'sleeping')
           AND "available_at" <= NOW()
           AND ("deadline_at" IS NULL OR "deadline_at" > NOW())
-          AND ("workflow_name" || '@' || COALESCE("version", '')) = ANY(${keys}::text[])
+          AND ("workflow_name" || '@' || COALESCE("version", '')) = ANY(${o.keys}::text[])
         ORDER BY CASE WHEN "status" = 'pending' THEN 0 ELSE 1 END, "available_at", "created_at"
         LIMIT 1
         FOR UPDATE SKIP LOCKED
@@ -101,7 +116,20 @@ function claimFilteredBackend(backend: BackendPostgres, defs: readonly TaskDefin
       FROM candidate
       WHERE wr."id" = candidate."id" AND wr."namespace_id" = ${b.namespaceId}
       RETURNING wr.*;`;
-      return claimed ?? null;
+    return claimed ?? null;
+  };
+  Object.defineProperty(filtered, "claimWorkflowRun", {
+    value: async (params: { workerId: string; leaseDurationMs: number }) => {
+      // 同步段先占位:同一轮 tick 的各槽依次进到这里,reserved 保证不超过动态上限
+      if (o.maxActive && o.active() + reserved >= Math.max(0, o.maxActive())) return null;
+      reserved += 1;
+      try {
+        if (o.gate && !(await o.gate())) return null;
+        return await claim(params);
+      } finally {
+        // 认领到的 run 在本函数返回后同步加入 worker 的执行集合,下一轮 tick 前已计入 active
+        reserved -= 1;
+      }
     },
   });
   return filtered;
@@ -189,16 +217,63 @@ export class OpenWorkflowEngine implements TaskEngine {
     return res.data.map((r) => toView(r, metas.get(r.id)));
   }
 
-  async startWorker(opts: StartWorkerOptions = {}) {
+  async startWorker(opts: StartWorkerOptions = {}): Promise<WorkerHandle> {
     const defs = this.opts.registry.select(opts.versions);
     const workflows = defs.map((d) => this.toWorkflow(d)) as Workflow<unknown, unknown, unknown>[];
     // 指定了 versions 的 worker 只认领这些任务:OpenWorkflow 0.10.1 认领不看任务名,认领到未注册的 run 会
-    // 以"未注册"失败并指数退避(5s→5m),分组 worker(io / agent)会互相拖慢对方的任务。
-    const backend = opts.versions?.length ? claimFilteredBackend(this.backend, defs) : this.backend;
-    const worker = new Worker({ backend, workflows, concurrency: opts.concurrency ?? 4 });
+    // 以"未注册"失败并指数退避(5s→5m),分组 worker(各道 / agent)会互相拖慢对方的任务。
+    const keys = opts.versions?.length ? [...new Set(defs.map((d) => `${d.name}@${engineVersion(d.version)}`))] : null;
+    let worker: Worker | null = null;
+    // biome-ignore lint/suspicious/noExplicitAny: OpenWorkflow Worker 的执行集合是私有字段,只读取其大小
+    const active = () => ((worker as any)?.activeExecutions?.size as number | undefined) ?? 0;
+    const backend =
+      keys || opts.maxActive || opts.claimGate
+        ? claimWrappedBackend(this.backend, {
+            keys,
+            active,
+            ...(opts.maxActive ? { maxActive: opts.maxActive } : {}),
+            ...(opts.claimGate ? { gate: opts.claimGate } : {}),
+          })
+        : this.backend;
+    const concurrency = opts.concurrency ?? 4;
+    worker = new Worker({ backend, workflows, concurrency });
     await worker.start();
     this.workers.push(worker);
     this.log.info("worker started", { tasks: defs.map((d) => `${d.name}@${d.version}`) });
+    return { concurrency, active };
+  }
+
+  /**
+   * 按任务名分组统计排队 / 运行数(全库):
+   * queued = 已到 available_at 可被认领(pending、等待后到点、租约过期待接手);
+   * running = worker 持有未过期租约。等信号 / sleep 未到点的不计。
+   */
+  async queueCounts(groups: readonly (readonly string[])[]): Promise<QueueCounts[]> {
+    // biome-ignore lint/suspicious/noExplicitAny: 同 claimWrappedBackend
+    const b = this.backend as any;
+    const table = b.workflowRunsTable();
+    const all = [...new Set(groups.flat())];
+    if (all.length === 0) return groups.map(() => ({ queued: 0, running: 0 }));
+    const rows: { name: string; queued: string | number; running: string | number }[] = await b.pg`
+      SELECT "workflow_name" AS name,
+        COUNT(*) FILTER (WHERE "available_at" <= NOW()) AS queued,
+        COUNT(*) FILTER (WHERE "status" = 'running' AND "worker_id" IS NOT NULL AND "available_at" > NOW()) AS running
+      FROM ${table}
+      WHERE "namespace_id" = ${b.namespaceId}
+        AND "status" IN ('pending', 'running', 'sleeping')
+        AND ("deadline_at" IS NULL OR "deadline_at" > NOW())
+        AND "workflow_name" = ANY(${all}::text[])
+      GROUP BY "workflow_name"`;
+    const by = new Map(rows.map((r) => [r.name, { queued: Number(r.queued), running: Number(r.running) }]));
+    return groups.map((names) =>
+      names.reduce(
+        (acc, n) => {
+          const c = by.get(n);
+          return c ? { queued: acc.queued + c.queued, running: acc.running + c.running } : acc;
+        },
+        { queued: 0, running: 0 },
+      ),
+    );
   }
 
   /** 等进行中的 handler 执行完再断开连接池 */
diff --git a/packages/tasks/src/registry.ts b/packages/tasks/src/registry.ts
index a2ecf09b..6ce4317d 100644
--- a/packages/tasks/src/registry.ts
+++ b/packages/tasks/src/registry.ts
@@ -2,10 +2,24 @@ import type { z } from "zod";
 import { UnknownTaskError } from "./errors.ts";
 import type { RetryPolicy, TaskContext } from "./types.ts";
 
+/**
+ * 执行道(WK-1):io worker 进程内按道分池,各道独立并发。
+ *   file        — 整理、硬链接/移动、ffprobe、校验等占磁盘的任务
+ *   network     — 站点/RSS/TMDB/下载器/搜索等外部请求
+ *   maintenance — 清理、回填、对账等可以等空闲再跑的任务(文件道/网络道有活时让路)
+ *   agent       — Agent 会话(worker:agent 进程,不在 io 进程的三条道里)
+ */
+export const TASK_LANES = ["file", "network", "maintenance", "agent"] as const;
+export type TaskLane = (typeof TASK_LANES)[number];
+/** 没声明 lane 的任务运行时归入网络道(启动时打 warn;ci 的归属表测试会直接失败) */
+export const DEFAULT_TASK_LANE: TaskLane = "network";
+
 export interface TaskDefinition<S extends z.ZodType = z.ZodType, O = unknown> {
   name: string;
   /** 正整数;不兼容改动 +1,并在 worker 上同时注册新旧版本直到旧 run 清空 */
   version: number;
+  /** 执行道;新任务必须显式声明(见 TASK_LANES) */
+  lane?: TaskLane;
   inputSchema: S;
   /** 步骤默认重试策略(步骤可单独覆盖)。外部网络类建议 5 次 1s→60s,文件类 3 次 */
   retryPolicy?: Partial<RetryPolicy>;
@@ -15,9 +29,14 @@ export interface TaskDefinition<S extends z.ZodType = z.ZodType, O = unknown> {
 export function defineTask<S extends z.ZodType, O>(def: TaskDefinition<S, O>): TaskDefinition<S, O> {
   if (!/^[A-Za-z][A-Za-z0-9_.-]*$/.test(def.name)) throw new Error(`invalid task name: ${def.name}`);
   if (!Number.isInteger(def.version) || def.version < 1) throw new Error(`invalid task version: ${def.version}`);
+  if (def.lane !== undefined && !(TASK_LANES as readonly string[]).includes(def.lane))
+    throw new Error(`invalid task lane: ${def.lane}`);
   return def;
 }
 
+/** 任务所属执行道;未声明时回退 DEFAULT_TASK_LANE */
+export const laneOf = (def: Pick<TaskDefinition, "lane">): TaskLane => def.lane ?? DEFAULT_TASK_LANE;
+
 export const engineVersion = (version: number) => `v${version}`;
 export const parseEngineVersion = (v: string | null): number | null => {
   const m = v ? /^v(\d+)$/.exec(v) : null;
diff --git a/packages/tasks/src/types.ts b/packages/tasks/src/types.ts
index deaaf18f..f012a1c7 100644
--- a/packages/tasks/src/types.ts
+++ b/packages/tasks/src/types.ts
@@ -56,7 +56,30 @@ export interface StartWorkerOptions {
    * 省略则执行注册表里的全部版本。升级期间新旧版本要同时列出。
    */
   versions?: string[];
+  /** 并发槽上限(引擎 worker 的槽位数,启动后不变);缺省 4 */
   concurrency?: number;
+  /**
+   * 动态并发:每次认领前读取,本 worker 执行中 + 认领中的 run 数达到它就不再认领(≤ concurrency 才有意义)。
+   * 用于设置页改并发后热生效(槽位按上限开,实际占用按它收)。
+   */
+  maxActive?: () => number;
+  /** 认领闸门:每次认领前调用,返回 false 本次不认领(维护道"空闲时跑"用) */
+  claimGate?: () => boolean | Promise<boolean>;
+}
+
+/** startWorker 返回的句柄:本 worker 的槽位与正在执行的 run 数 */
+export interface WorkerHandle {
+  readonly concurrency: number;
+  /** 本进程内该 worker 正在执行的 run 数 */
+  active(): number;
+}
+
+/** 一组任务在引擎里的排队 / 运行数(全库,跨进程) */
+export interface QueueCounts {
+  /** 已到可执行时间、等待认领的 run(含租约过期待接手的) */
+  queued: number;
+  /** 被某个 worker 持有租约、正在执行的 run */
+  running: number;
 }
 
 export interface TaskEngine {
@@ -70,7 +93,10 @@ export interface TaskEngine {
   cancel(runId: string, reason?: string): Promise<boolean>;
   get(runId: string): Promise<TaskRunView | null>;
   list(filter?: ListFilter): Promise<TaskRunView[]>;
-  startWorker(opts?: StartWorkerOptions): Promise<void>;
+  // biome-ignore lint/suspicious/noConfusingVoidType: 测试替身常写成 async startWorker() {}(返回 void)
+  startWorker(opts?: StartWorkerOptions): Promise<WorkerHandle | void>;
+  /** 按任务名分组统计排队 / 运行数(每组一条);不支持的实现可省略 */
+  queueCounts?(groups: readonly (readonly string[])[]): Promise<QueueCounts[]>;
   stop(): Promise<void>;
 }