diff --git a/.github/scripts/reuse.mjs b/.github/scripts/reuse.mjs
new file mode 100644
index 000000000..388060a70
--- /dev/null
+++ b/.github/scripts/reuse.mjs
@@ -0,0 +1,80 @@
+/** 只复用相同工作流对相同合并代码的完整成功检查；证据不足时执行全量门禁。 */
+export async function reuse({ github, context, core }) {
+  core.setOutput('reuse', 'false')
+  if (
+    context.eventName !== 'push' ||
+    context.payload.forced ||
+    !context.payload.before?.match(/^[a-f0-9]{40}$/) ||
+    /^0+$/.test(context.payload.before)
+  )
+    return
+
+  try {
+    const repo = context.repo
+    const { data: current } = await github.rest.actions.getWorkflowRun({ ...repo, run_id: context.runId })
+    const { data: pushed } = await github.rest.git.getCommit({ ...repo, commit_sha: context.sha })
+    const pulls = await github.paginate(github.rest.repos.listPullRequestsAssociatedWithCommit, {
+      ...repo,
+      commit_sha: context.sha,
+      per_page: 100,
+    })
+    for (const pull of pulls) {
+      if (
+        !pull.merged_at ||
+        pull.merge_commit_sha !== context.sha ||
+        pull.base?.ref !== context.ref.replace('refs/heads/', '') ||
+        pull.base?.repo?.full_name !== `${repo.owner}/${repo.repo}`
+      )
+        continue
+
+      // 不过滤 success：同一 PR 最新执行失败或仍在运行时，不能退回旧的绿灯。
+      const { data } = await github.rest.actions.listWorkflowRuns({
+        ...repo,
+        workflow_id: current.workflow_id,
+        event: 'pull_request',
+        head_sha: pull.head.sha,
+        per_page: 100,
+      })
+      const run = data.workflow_runs
+        .filter(
+          candidate => candidate.head_sha === pull.head.sha && candidate.head_repository?.id === pull.head.repo?.id,
+        )
+        .sort((left, right) => right.id - left.id)[0]
+      if (!run || run.status !== 'completed' || run.conclusion !== 'success') continue
+
+      const jobs = await github.paginate(github.rest.actions.listJobsForWorkflowRun, {
+        ...repo,
+        run_id: run.id,
+        filter: 'latest',
+        per_page: 100,
+      })
+      const proofs = jobs.filter(
+        job =>
+          job.status === 'completed' && job.conclusion === 'success' && /^CI proof \([a-f0-9]{40}\)$/.test(job.name),
+      )
+      if (proofs.length !== 1) continue
+      const testedSha = proofs[0].name.slice(10, -1)
+      const { data: tested } = await github.rest.git.getCommit({ ...repo, commit_sha: testedSha })
+      // run.head_sha 是 PR 分支头；只有 proof 记录的 github.sha 才是实际测试的模拟合并提交。
+      if (
+        tested.parents.length !== 2 ||
+        tested.parents[0].sha !== context.payload.before ||
+        tested.parents[1].sha !== pull.head.sha ||
+        tested.tree.sha !== pushed.tree.sha
+      )
+        continue
+
+      // 查询证明期间可能有人重新运行 CI，不能复用已经失效的运行快照。
+      const { data: latest } = await github.rest.actions.getWorkflowRun({ ...repo, run_id: run.id })
+      if (latest.status !== 'completed' || latest.conclusion !== 'success' || latest.run_attempt !== run.run_attempt)
+        continue
+
+      core.setOutput('reuse', 'true')
+      core.notice(`复用 PR #${pull.number} 的完整检查：${run.html_url}；代码树 ${tested.tree.sha}`)
+      return
+    }
+    core.info('没有匹配的完整 PR 检查，执行全量门禁。')
+  } catch (error) {
+    core.warning(`无法确认 PR 检查，执行全量门禁：${error.message}`)
+  }
+}
diff --git a/.github/scripts/reuse.test.mjs b/.github/scripts/reuse.test.mjs
new file mode 100644
index 000000000..8d113f25d
--- /dev/null
+++ b/.github/scripts/reuse.test.mjs
@@ -0,0 +1,171 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import { reuse } from './reuse.mjs'
+
+/** 构造带真实 GitHub 字段形状的离线 API 替身。 */
+function fixture() {
+  const head = 'a'.repeat(40),
+    base = 'b'.repeat(40),
+    sha = 'c'.repeat(40),
+    merge = 'd'.repeat(40)
+  const state = {
+    context: {
+      eventName: 'push',
+      repo: { owner: 'owner', repo: 'repo' },
+      ref: 'refs/heads/v3',
+      sha,
+      runId: 20,
+      payload: { before: base },
+    },
+    pull: {
+      number: 7,
+      merged_at: '2026-09-09T00:00:00Z',
+      merge_commit_sha: sha,
+      head: { sha: head, repo: { id: 2 } },
+      base: { ref: 'v3', repo: { full_name: 'owner/repo' } },
+    },
+    runs: [
+      {
+        id: 10,
+        run_attempt: 1,
+        status: 'completed',
+        conclusion: 'success',
+        head_sha: head,
+        head_repository: { id: 2 },
+        html_url: 'https://github.com/owner/repo/actions/runs/10',
+      },
+    ],
+    jobs: [{ name: `CI proof (${merge})`, status: 'completed', conclusion: 'success' }],
+    tested: { parents: [{ sha: base }, { sha: head }], tree: { sha: 'tree' } },
+    outputs: {},
+    calls: [],
+    warnings: [],
+  }
+  /** 保留 API 参数供测试验证，避免错误地过滤历史失败记录。 */
+  const api = name => async args => {
+    state.calls.push({ name, args })
+    if (state.error) throw new Error('API unavailable')
+    if (name === 'getWorkflowRun')
+      return { data: args.run_id === 20 ? { workflow_id: 100 } : { ...state.runs[0], ...state.refreshed } }
+    if (name === 'getCommit') return { data: args.commit_sha === sha ? { tree: { sha: 'tree' } } : state.tested }
+    if (name === 'listPullRequestsAssociatedWithCommit') return [state.pull]
+    if (name === 'listWorkflowRuns') return { data: { workflow_runs: state.runs } }
+    if (name === 'listJobsForWorkflowRun') return state.jobs
+    throw new Error(name)
+  }
+  state.github = {
+    rest: {
+      actions: {
+        getWorkflowRun: api('getWorkflowRun'),
+        listWorkflowRuns: api('listWorkflowRuns'),
+        listJobsForWorkflowRun: api('listJobsForWorkflowRun'),
+      },
+      git: { getCommit: api('getCommit') },
+      repos: { listPullRequestsAssociatedWithCommit: api('listPullRequestsAssociatedWithCommit') },
+    },
+    paginate: (method, args) => method(args),
+  }
+  state.core = {
+    setOutput: (key, value) => {
+      state.outputs[key] = value
+    },
+    notice: () => {},
+    info: () => {},
+    warning: message => state.warnings.push(message),
+  }
+  return state
+}
+
+test('相同合并代码且完整成功时复用，即使 fork run 没有 pull_requests 字段', async () => {
+  const state = fixture()
+  await reuse(state)
+  assert.equal(state.outputs.reuse, 'true')
+  const query = state.calls.find(call => call.name === 'listWorkflowRuns').args
+  assert.equal(query.workflow_id, 100)
+  assert.equal(query.event, 'pull_request')
+  assert.equal(query.status, undefined)
+  assert.equal(state.calls.find(call => call.name === 'listJobsForWorkflowRun').args.filter, 'latest')
+})
+
+const cases = {
+  'PR 必须完整执行': state => {
+    state.context.eventName = 'pull_request'
+  },
+  '手工触发必须完整执行': state => {
+    state.context.eventName = 'workflow_dispatch'
+  },
+  'force push 必须完整执行': state => {
+    state.context.payload.forced = true
+  },
+  '新建分支必须完整执行': state => {
+    state.context.payload.before = '0'.repeat(40)
+  },
+  '直接 push 没有对应合并': state => {
+    state.pull.merge_commit_sha = 'other'
+  },
+  '未合并 PR 不可复用': state => {
+    state.pull.merged_at = null
+  },
+  '目标分支不同': state => {
+    state.pull.base.ref = 'other'
+  },
+  '目标仓库不同': state => {
+    state.pull.base.repo.full_name = 'other/repo'
+  },
+  'PR head 已改变': state => {
+    state.pull.head.sha = 'e'.repeat(40)
+  },
+  'PR 来源仓库不同': state => {
+    state.pull.head.repo.id = 9
+  },
+  '旧工作流没有 proof': state => {
+    state.jobs = []
+  },
+  'proof 被跳过': state => {
+    state.jobs[0].conclusion = 'skipped'
+  },
+  'proof 尚未完成': state => {
+    state.jobs[0].status = 'in_progress'
+  },
+  'proof 重复': state => {
+    state.jobs.push({ ...state.jobs[0] })
+  },
+  '工作流有失败': state => {
+    state.runs[0].conclusion = 'failure'
+  },
+  '工作流尚未完成': state => {
+    state.runs[0].status = 'in_progress'
+  },
+  '最新执行失败不可退回旧成功': state => {
+    state.runs.push({ ...state.runs[0], id: 11, conclusion: 'failure' })
+  },
+  '查询期间开始重跑': state => {
+    state.refreshed = { status: 'in_progress', conclusion: null }
+  },
+  '查询期间运行轮次变化': state => {
+    state.refreshed = { run_attempt: 2 }
+  },
+  '目标分支已经前进': state => {
+    state.tested.parents[0].sha = 'e'.repeat(40)
+  },
+  '证明不是 PR 模拟合并': state => {
+    state.tested.parents.pop()
+  },
+  '模拟合并来自其他 PR head': state => {
+    state.tested.parents[1].sha = 'e'.repeat(40)
+  },
+  '合入代码树有变化': state => {
+    state.tested.tree.sha = 'different'
+  },
+  'API 异常回退全量': state => {
+    state.error = true
+  },
+}
+for (const [name, mutate] of Object.entries(cases)) {
+  test(name, async () => {
+    const state = fixture()
+    mutate(state)
+    await reuse(state)
+    assert.equal(state.outputs.reuse, 'false')
+  })
+}
diff --git a/.github/workflows/pylint.yml b/.github/workflows/pylint.yml
index fb77b623d..9d77face9 100644
--- a/.github/workflows/pylint.yml
+++ b/.github/workflows/pylint.yml
@@ -17,7 +17,32 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
+  reuse:
+    name: Check reusable PR verification
+    runs-on: ubuntu-latest
+    timeout-minutes: 3
+    permissions:
+      contents: read
+      actions: read
+      pull-requests: read
+    outputs:
+      reused: ${{ steps.check.outputs.reuse }}
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+      - name: Test reuse decision
+        run: node --test .github/scripts/reuse.test.mjs
+      - name: Check successful PR merge tree
+        id: check
+        uses: actions/github-script@v9
+        with:
+          script: |
+            const { reuse } = await import(`${process.env.GITHUB_WORKSPACE}/.github/scripts/reuse.mjs`);
+            await reuse({ github, context, core });
+
   pylint:
+    needs: reuse
+    if: needs.reuse.outputs.reused != 'true'
     runs-on: ubuntu-latest
     name: Pylint Code Quality Check
 
@@ -105,3 +130,13 @@ jobs:
         echo "🎉 Pylint 检查完成！"
         echo "✅ 改动 Python 文件没有新增语法错误或严重问题"
         echo "📊 全仓建议性报告已保存为构建工件"
+
+  proof:
+    name: CI proof (${{ github.sha }})
+    needs: [pylint]
+    if: github.event_name == 'pull_request' && needs.pylint.result == 'success'
+    runs-on: ubuntu-latest
+    timeout-minutes: 1
+    steps:
+      - name: Record fully verified merge commit
+        run: echo 'All required checks passed for this PR merge commit.'
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index d12e9cac4..0f58be451 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -1,7 +1,7 @@
 name: Unit Tests
 
 on:
-  # 指向 v3 的 PR 与推送都跑全量单测，作为合并门禁
+  # PR 完整验证；合并推送仅在代码树和成功证明匹配时复用
   pull_request:
     branches:
       - v3
@@ -19,7 +19,32 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
+  reuse:
+    name: Check reusable PR verification
+    runs-on: ubuntu-latest
+    timeout-minutes: 3
+    permissions:
+      contents: read
+      actions: read
+      pull-requests: read
+    outputs:
+      reused: ${{ steps.check.outputs.reuse }}
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+      - name: Test reuse decision
+        run: node --test .github/scripts/reuse.test.mjs
+      - name: Check successful PR merge tree
+        id: check
+        uses: actions/github-script@v9
+        with:
+          script: |
+            const { reuse } = await import(`${process.env.GITHUB_WORKSPACE}/.github/scripts/reuse.mjs`);
+            await reuse({ github, context, core });
+
   architecture:
+    needs: reuse
+    if: needs.reuse.outputs.reused != 'true'
     runs-on: ubuntu-latest
     name: Architecture Contract Gate
     timeout-minutes: 10
@@ -90,42 +115,11 @@ jobs:
           uv run --locked --no-sync python
           scripts/startup/performance.py --check --repeat 3
 
-  pytest:
-    runs-on: ubuntu-latest
-    name: Unit Tests (${{ matrix.shard }})
-    timeout-minutes: 20
-    strategy:
-      fail-fast: false
-      matrix:
-        include:
-          - shard: '1/4'
-          - shard: '2/4'
-          - shard: '3/4'
-          - shard: '4/4'
-
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v7
-
-      - name: Set up uv
-        uses: astral-sh/setup-uv@v10.0.1
-        with:
-          python-version: '3.14'
-          enable-cache: true
-          cache-dependency-glob: |
-            pyproject.toml
-            uv.lock
-
-      - name: Install dependencies
-        run: uv sync --locked
-
-      - name: Run tests
-        timeout-minutes: 10
-        run: uv run --locked --no-sync python tests/run.py --shard "${{ matrix.shard }}"
-
   coverage-shard:
+    needs: reuse
+    if: needs.reuse.outputs.reused != 'true'
     runs-on: ubuntu-latest
-    name: Coverage Shard (${{ matrix.shard }})
+    name: Unit Tests with Coverage (${{ matrix.shard }})
     timeout-minutes: 15
     strategy:
       fail-fast: false
@@ -222,3 +216,13 @@ jobs:
             coverage.xml
             coverage.json
           retention-days: 7
+
+  proof:
+    name: CI proof (${{ github.sha }})
+    needs: [architecture, coverage-shard, coverage-report]
+    if: github.event_name == 'pull_request' && needs.architecture.result == 'success' && needs.coverage-shard.result == 'success' && needs.coverage-report.result == 'success'
+    runs-on: ubuntu-latest
+    timeout-minutes: 1
+    steps:
+      - name: Record fully verified merge commit
+        run: echo 'All required checks passed for this PR merge commit.'
diff --git a/AGENTS.md b/AGENTS.md
index 6db191c2a..b52f1dfd2 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -42,7 +42,7 @@ For work that changes or reviews repository behavior, identify the domains actua
 
 ### Testing
 * **Primary Reference:** `docs/testing.md`
-* **Required Constraints:** pytest is the only runner; `tests/conftest.py` isolates each run to a temporary `CONFIG_DIR`. Tests must not touch the real database, network, or external services (TMDB, LLM catalogs, downloaders, media servers, MP server) — mock at the boundary or replay recorded responses; the bar is zero real outbound traffic. Tests must restore any process-level state they stub (`sys.modules`, singletons, caches, settings). New tests must be pytest-native (function + `assert` + fixtures); do not add new `unittest.TestCase`. Convert existing `TestCase` files to pytest-native opportunistically when you modify them. Before opening a PR to `v3`, run the affected tests and applicable local checks. Run the full local suite (`uv run --locked --no-sync python tests/run.py`) for dependency or lock changes, shared test infrastructure, database or startup paths, cross-module lifecycle, compatibility layers, broad behavior changes, or an explicit maintainer requirement. The changed path must pass; any unrelated failure must be reported and reproduced against the current `upstream/v3` baseline instead of silently expanding the PR. Documentation-only changes use applicable text and structure checks; the `.github/workflows/test.yml` gate remains the final full-suite check on every PR/push to `v3`.
+* **Required Constraints:** pytest is the only runner; `tests/conftest.py` isolates each run to a temporary `CONFIG_DIR`. Tests must not touch the real database, network, or external services (TMDB, LLM catalogs, downloaders, media servers, MP server) — mock at the boundary or replay recorded responses; the bar is zero real outbound traffic. Tests must restore any process-level state they stub (`sys.modules`, singletons, caches, settings). New tests must be pytest-native (function + `assert` + fixtures); do not add new `unittest.TestCase`. Convert existing `TestCase` files to pytest-native opportunistically when you modify them. Before opening a PR to `v3`, run the affected tests and applicable local checks. Run the full local suite (`uv run --locked --no-sync python tests/run.py`) for dependency or lock changes, shared test infrastructure, database or startup paths, cross-module lifecycle, compatibility layers, broad behavior changes, or an explicit maintainer requirement. The changed path must pass; any unrelated failure must be reported and reproduced against the current `upstream/v3` baseline instead of silently expanding the PR. Documentation-only changes use applicable text and structure checks; the `.github/workflows/test.yml` gate verifies every PR/push to `v3`; a merge push may reuse a complete successful PR verification only when the tested merge tree and base match exactly (see `docs/testing.md`).
 
 ### Commands and Development Workflow
 * **Primary Reference:** `docs/rules/03-commands.md`
diff --git a/docs/testing.md b/docs/testing.md
index 509b65ded..8c0e00b42 100644
--- a/docs/testing.md
+++ b/docs/testing.md
@@ -10,14 +10,19 @@ pytest 是唯一运行入口。`tests/conftest.py` 在收集前完成隔离引
 uv run --locked --no-sync pytest tests                              # 串行全量
 uv run --locked --no-sync pytest tests/test_xxx.py                  # 单文件
 uv run --locked --no-sync pytest tests/test_xxx.py::SomeTest::test_y   # 单用例
-uv run --locked --no-sync python tests/run.py                       # 默认按文件连续切成 4 片并行跑全量
+uv run --locked --no-sync python tests/run.py                       # 默认按文件预计耗时均衡为 4 片并行跑全量
 uv run --locked --no-sync python tests/run.py --serial              # 串行全量，便于调试或生成覆盖率
 uv run --locked --no-sync python tests/run.py --shard 1/4           # 只跑指定分片，供 CI 复用
 ```
 
 `tests/run.py` 的 runner 参数只有 `--serial` 和 `--shard N/TOTAL`；其余参数保持原顺序
-透传给 pytest，例如 `python tests/run.py -q --maxfail=1`。文件先按字典序排序，再以
-`ceil(文件数 / 分片数)` 的大小连续切片，确保本地与 CI 执行相同的文件集合和顺序。
+透传给 pytest，例如 `python tests/run.py -q --maxfail=1`。分片使用受版本控制的
+`tests/fixtures/durations.json` 慢文件耗时估算：按耗时降序，逐个分给预计总耗时最短的分片；
+耗时相同时按路径、分片编号决定归属，片内仍按路径排序。新增或未记录文件使用 1 秒权重，
+不依赖本机缓存，确保同一代码版本在本地与 CI 的文件集合和顺序一致。
+耗时数据来自成功 Ubuntu/Python 3.14 Coverage 日志的逐文件进度时间戳（含续行），
+只记录至少 3 秒的文件并向上取整；后续发现慢文件偏移时可依据成功 CI 日志更新。
+这些估算仅用于调度，不构成性能或覆盖率基线。
 
 - 不再使用 `python -m unittest discover`：它不导入 `tests` 包、收不到纯函数用例，且绕过 `conftest.py` 的隔离。
 - 不再依赖 `python tests/test_xxx.py` 直跑：所有 `if __name__ == "__main__": unittest.main()` 尾巴已移除。
@@ -156,9 +161,12 @@ def test_recognize_prefers_explicit_identity(sample_meta, monkeypatch):
 
 ## CI 与 PR
 
-- **门禁**：`.github/workflows/test.yml` 在指向 `v3` 的 `pull_request` / `push` 及手动触发时，从 `uv.lock` 同步环境。独立 `architecture` job 先运行宿主依赖、运行契约和基线 CLI 快速门禁；全量测试再通过 `tests/run.py --shard N/TOTAL` 稳定分到 4 个 pytest job。每个分片都有独立进程和临时 `CONFIG_DIR`，不共用 SQLite 或进程级状态。Coverage 另以 8 个并行分片采集数据，由单一报告 job 合并后检查 Application 与 Domain 的固定 80% 基线。
+- **合并检查复用**：单测/架构与 Pylint 工作流各自保留 PR、push 和手动入口。PR 完整通过全部门禁后，末尾 `CI proof (<github.sha>)` job 记录实际验证的模拟合并提交。合并 push 只在同一工作流的最新 PR 运行完整成功、证明成功、代码树完全相同、模拟合并父提交分别等于 push 前的目标分支和 PR head 时跳过重复门禁。直接 push、强制 push、基线变化、旧工作流缺少证明、失败/未完成运行或 API 异常都执行全量；不依赖提交消息，也不把 PR 的 head SHA 当作测试的合并 SHA。标准 merge/squash/rebase 仅在上述证据一致时复用。GitHub 构建与发布工作流保持独立。
+- **去重脚本验证**：`node --test .github/scripts/reuse.test.mjs` 离线覆盖成功复用及保守回退，两个检查工作流均在判定前执行。复用 job 仅持有 contents/actions/pull-requests 读取权限；没有修改分支保护设置。
+
+- **门禁**：`.github/workflows/test.yml` 在指向 `v3` 的 `pull_request` / `push` 及手动触发时，从 `uv.lock` 同步环境。独立 `architecture` job 先运行宿主依赖、运行契约和基线 CLI 快速门禁；全量测试通过 `coverage run --parallel-mode tests/run.py --shard N/8` 分到 8 个 job，一次执行同时验证单测并采集覆盖率。每个分片都有独立进程和临时 `CONFIG_DIR`，不共用 SQLite 或进程级状态，由单一报告 job 合并后检查 Application 与 Domain 的固定 80% 基线。
 - **跨仓观察**：`.github/workflows/architecture-observe.yml` 每周或手工检出官方插件仓最新 `main`，使用 `--check-plugins` 比较公开导入、Hook 和动态 API 契约。它只上传 `official-plugin-architecture-report.json`，不会自动刷新 fixture；语义变化必须人工审查后显式执行 `--write-plugins`。
 - **静态检查**：`.github/workflows/pylint.yml` 对指向 `v3` 的 PR、推送和手工触发运行 Pylint。PR/推送改动到的 Python 文件是硬门禁；`app/` 全量扫描保留为建议性 JSON 构建工件，存量告警不会掩盖或阻塞本次增量治理。
 - **PR 本地验证**：提交前运行受影响测试和适用的静态检查。涉及依赖或锁文件、共享测试基建、数据库、启动链、跨模块生命周期、兼容层或大范围行为变化时，运行 `uv run --locked --no-sync python tests/run.py` 完成本地全量；需要断点、输出顺序或测试污染诊断时使用 `--serial`。所有测试都应确认受影响路径通过且 socket 探针无真实出站，验证说明准确标注执行范围。若存在无关失败，必须在当前 `upstream/v3` 基线上独立复现并在 PR 中如实说明；不得静默扩大当前 PR 去修复基线问题。纯文档变更执行适用的文本、结构和 diff 检查，CI 继续运行全量门禁。
-- **覆盖率门禁**：`Coverage Shard` jobs 会在 `v3` 的 PR、push 和手工触发中通过 `tests/run.py --shard N/8` 并行采集覆盖率数据，`Coverage Report` 再合并全部分片并只读检查 Application 与 Domain 是否达到 Ubuntu/Python 3.14 canonical 的固定 80% 行覆盖率基线，同时上传 JSON / XML 工件。低于 80% 会阻塞；达到或超过 80% 不要求同步运行时语句计数。macOS 本地报告只用于诊断，不直接作为可提交基线。
+- **覆盖率门禁**：`Unit Tests with Coverage` jobs 会在 `v3` 的 PR、push 和手工触发中通过 `tests/run.py --shard N/8` 并行采集覆盖率数据，`Coverage Report` 再合并全部分片并只读检查 Application 与 Domain 是否达到 Ubuntu/Python 3.14 canonical 的固定 80% 行覆盖率基线，同时上传 JSON / XML 工件。低于 80% 会阻塞；达到或超过 80% 不要求同步运行时语句计数。macOS 本地报告只用于诊断，不直接作为可提交基线。
 - 复现 CI 使用 `uv sync --locked`；主程序运行依赖位于 `[project].dependencies`，pytest 与覆盖率工具位于默认 `dev` 依赖组。
diff --git a/tests/fixtures/durations.json b/tests/fixtures/durations.json
new file mode 100644
index 000000000..71dfafd01
--- /dev/null
+++ b/tests/fixtures/durations.json
@@ -0,0 +1,37 @@
+{
+  "source": "https://github.com/jxxghp/MoviePilot/actions/runs/34326739404",
+  "description": "Ubuntu/Python 3.14 coverage 日志中逐文件进度时间戳的耗时估算，含续行；仅记录至少 3 秒的文件并向上取整，其余文件使用 1 秒权重。用于分片均衡，不是性能或覆盖率基线。",
+  "files": {
+    "test_agent_api_lazy_imports.py": 15,
+    "test_agent_api_surface_audit.py": 5,
+    "test_agent_lazy_runtime_boundary.py": 7,
+    "test_agent_tool_result_policy.py": 4,
+    "test_api_response.py": 29,
+    "test_architecture_adapter_imports.py": 67,
+    "test_architecture_baseline_cli.py": 7,
+    "test_architecture_contract_baseline.py": 6,
+    "test_architecture_dependencies.py": 95,
+    "test_architecture_egress.py": 67,
+    "test_architecture_event_facts.py": 17,
+    "test_architecture_event_policy.py": 16,
+    "test_bangumi_media_type.py": 4,
+    "test_chain_base_boundary.py": 9,
+    "test_data_cleanup_chain.py": 7,
+    "test_duplicate_code.py": 6,
+    "test_execute_command_tool.py": 4,
+    "test_host_runtime_context.py": 4,
+    "test_legacy_import_compat.py": 5,
+    "test_llm_facade_governance.py": 4,
+    "test_main_direct_execution.py": 4,
+    "test_media_auxiliary.py": 4,
+    "test_module_manager_capability_adapter.py": 36,
+    "test_monitor_mount_isolation.py": 5,
+    "test_music_matching.py": 6,
+    "test_router_aggregation.py": 10,
+    "test_service_locator_gate.py": 5,
+    "test_subscription_governance_scale.py": 6,
+    "test_subscription_search_governance.py": 4,
+    "test_task_ownership_gate.py": 4,
+    "test_transfer_history_architecture.py": 6
+  }
+}
diff --git a/tests/run.py b/tests/run.py
index 56e5617aa..7940cdf00 100644
--- a/tests/run.py
+++ b/tests/run.py
@@ -2,16 +2,19 @@
 from __future__ import annotations
 
 import argparse
+import json
 import subprocess
 import sys
 from pathlib import Path
-from typing import Sequence
+from typing import Mapping, Sequence
 
 import pytest
 
 TESTS_DIR = Path(__file__).resolve().parent
 RUNNER_PATH = Path(__file__).resolve()
 DEFAULT_SHARD_COUNT = 4
+DURATIONS_PATH = TESTS_DIR / "fixtures" / "durations.json"
+DEFAULT_TEST_DURATION = 1.0
 
 
 def collect_test_files() -> list[Path]:
@@ -19,20 +22,30 @@ def collect_test_files() -> list[Path]:
     return sorted(TESTS_DIR.glob("test_*.py"))
 
 
+def load_test_durations() -> dict[str, float]:
+    """读取受版本控制的 CI 慢文件耗时估算，避免依赖本机缓存改变分片。"""
+    return json.loads(DURATIONS_PATH.read_text(encoding="utf-8"))["files"]
+
+
 def split_test_files(
-    test_files: Sequence[Path], shard_count: int
+    test_files: Sequence[Path], shard_count: int,
+    durations: Mapping[str, float] | None = None,
 ) -> list[list[Path]]:
-    """把排序后的文件连续均分，保持 CI 分片归属稳定且易于复现。"""
+    """把慢文件优先分给预计耗时最短的分片，并保持路径和平局选择确定。"""
     if shard_count <= 0:
         raise ValueError("shard_count 必须大于 0")
-    shard_size = (len(test_files) + shard_count - 1) // shard_count
-    if shard_size == 0:
-        return [[] for _ in range(shard_count)]
-    shards = [
-        list(test_files[start:start + shard_size])
-        for start in range(0, len(test_files), shard_size)
-    ]
-    return shards + [[] for _ in range(shard_count - len(shards))]
+    durations = load_test_durations() if durations is None else durations
+    shards: list[list[Path]] = [[] for _ in range(shard_count)]
+    shard_durations = [0.0] * shard_count
+    weighted_files = sorted(
+        test_files,
+        key=lambda path: (-durations.get(path.name, DEFAULT_TEST_DURATION), path),
+    )
+    for test_file in weighted_files:
+        shard_index = min(range(shard_count), key=shard_durations.__getitem__)
+        shards[shard_index].append(test_file)
+        shard_durations[shard_index] += durations.get(test_file.name, DEFAULT_TEST_DURATION)
+    return [sorted(shard) for shard in shards]
 
 
 def parse_shard(value: str) -> tuple[int, int]:
diff --git a/tests/test_architecture_ci.py b/tests/test_architecture_ci.py
index 7421ec815..a61b26f6c 100644
--- a/tests/test_architecture_ci.py
+++ b/tests/test_architecture_ci.py
@@ -86,6 +86,7 @@ def test_official_plugin_observation_is_scheduled_and_never_writes_fixture():
 def test_coverage_jobs_parallelize_data_and_keep_one_global_ratchet() -> None:
     """Coverage 分片并行采集数据，再由单一报告 job 合并并检查全局低水位。"""
     workflow = _load_workflow("test.yml")
+    assert "pytest" not in workflow["jobs"]
     shard_job = workflow["jobs"]["coverage-shard"]
     shard_steps = shard_job["steps"]
     report_job = workflow["jobs"]["coverage-report"]
@@ -94,6 +95,7 @@ def test_coverage_jobs_parallelize_data_and_keep_one_global_ratchet() -> None:
     assert workflow["on"]["push"]["branches"] == ["v3"]
     assert "workflow_dispatch" in workflow["on"]
     assert shard_job["runs-on"] == "ubuntu-latest"
+    assert shard_job["name"] == "Unit Tests with Coverage (${{ matrix.shard }})"
     assert shard_job["timeout-minutes"] == 15
     shard_matrix = shard_job["strategy"]["matrix"]["include"]
     assert [item["shard"] for item in shard_matrix] == [
@@ -170,6 +172,34 @@ def test_coverage_jobs_parallelize_data_and_keep_one_global_ratchet() -> None:
     assert "app/plugins/*/*" in coverage_config["run"]["omit"].splitlines()
 
 
+def test_ci_reuse_requires_successful_proof_for_every_gate() -> None:
+    """证明必须依赖全部硬门禁，推送只能通过保守判定跳过昂贵任务。"""
+    for filename, gates in {
+        "test.yml": ["architecture", "coverage-shard", "coverage-report"],
+        "pylint.yml": ["pylint"],
+    }.items():
+        jobs = _load_workflow(filename)["jobs"]
+        reuse = jobs["reuse"]
+        assert reuse["permissions"] == {
+            "contents": "read", "actions": "read", "pull-requests": "read"
+        }
+        assert reuse["outputs"]["reused"] == "${{ steps.check.outputs.reuse }}"
+        assert "node --test .github/scripts/reuse.test.mjs" in _step_commands(
+            _load_workflow(filename), "reuse"
+        )
+        for gate in gates:
+            if gate == "coverage-report":
+                continue
+            assert jobs[gate]["needs"] == "reuse"
+            assert jobs[gate]["if"] == "needs.reuse.outputs.reused != 'true'"
+        proof = jobs["proof"]
+        assert proof["name"] == "CI proof (${{ github.sha }})"
+        assert proof["needs"] == gates
+        assert proof["if"] == "github.event_name == 'pull_request' && " + " && ".join(
+            f"needs.{gate}.result == 'success'" for gate in gates
+        )
+
+
 def test_pylint_workflow_runs_for_v3_pull_requests_and_pushes():
     """改动文件应硬门禁，而全仓存量问题只能生成建议性报告。"""
     workflow = _load_workflow("pylint.yml")
diff --git a/tests/test_test_runner.py b/tests/test_test_runner.py
index e998b4941..43e51e917 100644
--- a/tests/test_test_runner.py
+++ b/tests/test_test_runner.py
@@ -16,14 +16,55 @@ def _test_files(count: int) -> list[Path]:
     return [Path(f"test_{index:03d}.py") for index in range(count)]
 
 
-def test_split_test_files_uses_stable_contiguous_chunks() -> None:
-    """文件分片必须稳定覆盖全集，且与既有 CI 的连续均分语义一致。"""
+def test_split_test_files_balances_slow_files_deterministically() -> None:
+    """相邻慢文件必须分散，输入顺序变化也不能造成分片漂移或遗漏。"""
     test_files = _test_files(10)
+    durations = {test_files[0].name: 12.0, test_files[1].name: 11.0}
 
-    shards = test_runner.split_test_files(test_files, shard_count=4)
+    shards = test_runner.split_test_files(test_files, shard_count=4, durations=durations)
 
-    assert [len(shard) for shard in shards] == [3, 3, 3, 1]
-    assert [test_file for shard in shards for test_file in shard] == test_files
+    assert shards == test_runner.split_test_files(
+        list(reversed(test_files)), shard_count=4, durations=durations,
+    )
+    assert sorted(test_file for shard in shards for test_file in shard) == test_files
+    assert all(shard == sorted(shard) for shard in shards)
+    assert not any(test_files[0] in shard and test_files[1] in shard for shard in shards)
+    assert max(sum(durations.get(path.name, 1.0) for path in shard) for shard in shards) == 12
+
+
+@pytest.mark.parametrize("file_count, shard_count", [(0, 4), (2, 4), (10, 4), (10, 1)])
+def test_split_test_files_covers_unknown_files_and_empty_shards(
+    file_count: int, shard_count: int,
+) -> None:
+    """新增无历史耗时文件和空分片均须保留，文件不得重复或漏跑。"""
+    test_files = _test_files(file_count)
+
+    shards = test_runner.split_test_files(test_files, shard_count, durations={})
+
+    assert len(shards) == shard_count
+    assert sorted(path for shard in shards for path in shard) == test_files
+    assert max(map(len, shards)) - min(map(len, shards)) <= 1
+
+
+def test_split_test_files_rejects_nonpositive_shard_count() -> None:
+    """分片数非法时须明确拒绝，不能静默丢失测试。"""
+    with pytest.raises(ValueError, match="shard_count"):
+        test_runner.split_test_files(_test_files(1), shard_count=0)
+
+
+def test_recorded_durations_balance_the_current_ci_suite() -> None:
+    """慢文件权重必须有效，并将当前全集的八分片预计耗时维持均衡。"""
+    durations = test_runner.load_test_durations()
+    assert all(value > 0 for value in durations.values())
+    test_files = test_runner.collect_test_files()
+    shards = test_runner.split_test_files(test_files, shard_count=8)
+    totals = [sum(durations.get(path.name, 1.0) for path in shard) for shard in shards]
+
+    assert sorted(path for shard in shards for path in shard) == test_files
+    assert max(totals) - min(totals) <= max(durations.values())
+    slow_files = sorted(durations, key=durations.get, reverse=True)[:3]
+    assert len({index for index, shard in enumerate(shards)
+                if any(path.name in slow_files for path in shard)}) == 3
 
 
 def test_main_defaults_to_four_parallel_shards(monkeypatch) -> None:
@@ -33,6 +74,7 @@ def test_main_defaults_to_four_parallel_shards(monkeypatch) -> None:
     monkeypatch.setattr(test_runner, "collect_test_files", lambda: test_files)
 
     def fake_run_parallel(shards, pytest_args):
+        """记录默认并行入口接收的分片与 pytest 参数。"""
         captured["shards"] = shards
         captured["pytest_args"] = pytest_args
         return 0
@@ -40,7 +82,7 @@ def test_main_defaults_to_four_parallel_shards(monkeypatch) -> None:
     monkeypatch.setattr(test_runner, "run_parallel_shards", fake_run_parallel)
 
     assert test_runner.main(["-q", "--maxfail=1"]) == 0
-    assert [len(shard) for shard in captured["shards"]] == [3, 3, 3, 1]
+    assert [len(shard) for shard in captured["shards"]] == [3, 3, 2, 2]
     assert captured["pytest_args"] == ["-q", "--maxfail=1"]
 
 
@@ -51,6 +93,7 @@ def test_main_runs_requested_ci_shard_in_current_process(monkeypatch) -> None:
     monkeypatch.setattr(test_runner, "collect_test_files", lambda: test_files)
 
     def fake_run_pytest(paths, pytest_args):
+        """记录 pytest 入口接收的文件与透传参数。"""
         captured["paths"] = paths
         captured["pytest_args"] = pytest_args
         return 0
@@ -58,7 +101,7 @@ def test_main_runs_requested_ci_shard_in_current_process(monkeypatch) -> None:
     monkeypatch.setattr(test_runner, "run_pytest", fake_run_pytest)
 
     assert test_runner.main(["--shard", "2/4", "-q"]) == 0
-    assert captured["paths"] == test_files[3:6]
+    assert captured["paths"] == test_files[1::4]
     assert captured["pytest_args"] == ["-q"]
 
 
@@ -67,6 +110,7 @@ def test_main_serial_preserves_legacy_full_suite_entry(monkeypatch) -> None:
     captured = {}
 
     def fake_run_pytest(paths, pytest_args):
+        """记录 pytest 入口接收的文件与透传参数。"""
         captured["paths"] = paths
         captured["pytest_args"] = pytest_args
         return 0
@@ -89,7 +133,7 @@ def test_workflow_uses_the_shared_runner_contract() -> None:
     """CI 不得另行维护 shell 分片算法，Coverage 必须复用同一分片入口。"""
     workflow = WORKFLOW.read_text(encoding="utf-8")
 
-    assert 'python tests/run.py --shard "${{ matrix.shard }}"' in workflow
+    assert 'run: uv run --locked --no-sync python tests/run.py --shard' not in workflow
     assert "python -m coverage run --parallel-mode tests/run.py --shard" in workflow
     assert "mapfile" not in workflow
     assert "SHARD_INDEX" not in workflow