feat(server): 同步引擎与 seed（阶段 4）

- 引擎：tick（5 分钟 + 防重入 + 启动即跑）· 单轮串行遍历 active key · 日志整批单事务入库（ON CONFLICT DO NOTHING + log_synced_at 取请求发起时刻）· 额度采集（上游字段全组，独立 try）· QPU 采样（基线 / 值变化才追加）· 每轮摘要与超期 warning
- 401 状态机：N=3 摘除（disabled_source=auto_401）；403/500/网络不计数；解密失败跳过不计数
- 纯逻辑抽层：policy（计数 / 合并判定）· dedupe（去重键与内容指纹）· log-row（日志映射，与 seed 共用）
- seed 脚本（幂等）：假 key + 984 条日志 + QPU 基线 + 示例购买
- 实测：401 端到端 1→2→3 摘除且摘除后不再轮询；幂等重跑新增 0 条
- 实证发现：错误行 request_id 不唯一（8 组 ×3）——记入实施计划「待定问题 #5」与 upstream 文档，待作者裁决
- AGENTS.md / 根 package.json 增 seed 命令

AGENTS.md

@@ -35,8 +35,9 @@ bun run type-check   # 类型检查（vue-tsc + tsc）
 bun run db:generate  # 生成 Drizzle 迁移（apps/server/drizzle）
 bun run db:migrate   # 执行迁移
 bun run db:push      # 直接推 schema（开发期）
+bun run seed         # 灌入开发数据（样例快照 → 四张表；幂等，仅本地）
 ```
 
-- 环境变量：复制 `.env.example` 为 `.env` 并填值（`DATABASE_URL` / `MANAGE_TOKEN` / `KEYS_MASTER_KEY` / `NEWAPI_BASE`）
+- 环境变量：复制 `.env.example` 为 `.env` 并填值（`DATABASE_URL` / `MANAGE_TOKEN` / `KEYS_MASTER_KEY` / `NEWAPI_BASE`；离线开发可加 `UPSTREAM_SOURCE=fixture` 走样例快照，见 `docs/design/实施计划.md` §6）
 - 目录速记：前端 `apps/web/src/{api,router,views}` · 后端 `apps/server/src/{db,routes,sync,lib}` · 契约类型在 `packages/shared`
 - 开发期前端 `/api` 由 Vite 代理到 :3000（`apps/web/vite.config.ts`）；生产由同域反代承担（见 README 工程注记）

apps/server/package.json

@@ -6,6 +6,7 @@
     "dev": "bun run --hot src/index.ts",
     "start": "bun run src/index.ts",
     "test": "bun test",
+    "seed": "bun run scripts/seed.ts",
     "db:generate": "drizzle-kit generate",
     "db:migrate": "drizzle-kit migrate",
     "db:push": "drizzle-kit push"

apps/server/scripts/seed.ts

@@ -0,0 +1,127 @@
+/**
+ * seed 脚本（实施计划 §6）：从仓库根的实测响应快照灌入开发数据——**幂等**，可重复执行。
+ *
+ * 用法：在 `apps/server` 下 `bun run seed`（环境变量自动经 `src/lib/env.ts` 加载仓库根 `.env`）。
+ *
+ * 灌入内容：
+ * 1. 一个占位 `api_keys` 行（假 key 常量 `SEED_KEY`，哈希 / 密文走真实 crypto 逻辑——
+ *    查询面可用该常量 key 命中已登记 key）
+ * 2. `usage_logs`：`api.log.token.json` 的 1000 条样例日志（`ON CONFLICT DO NOTHING`）
+ * 3. `qpu_history` 基线：`api.status.json` 的 `quota_per_unit`（已有数据则跳过）
+ * 4. 一条示例 `purchases`（订阅路径；**金额为占位值**，用于验证折算链）
+ *
+ * 注：快照文件未入库（`.gitignore` 的 `/api*.json`），本脚本只在本地开发环境使用。
+ */
+import { readFileSync } from 'node:fs'
+import { eq } from 'drizzle-orm'
+
+import { DEFAULT_QUOTA_PER_UNIT } from '@naku/shared'
+
+import { getDb } from '../src/db/client'
+import { apiKeys, purchases, qpuHistory, usageLogs } from '../src/db/schema'
+import { getKeySecrets } from '../src/lib/keys'
+import { fromUnixSeconds } from '../src/lib/time'
+import { parseOther, toLogRow } from '../src/sync/log-row'
+import type { UpstreamLogRecord } from '../src/upstream/types'
+
+/** 假 key 常量：查看面 / 管理面用它做本地验证（可用 env SEED_KEY 覆盖） */
+const SEED_KEY = process.env.SEED_KEY ?? 'sk-seed-demo-key-0001'
+
+/** 仓库根（本文件位于 apps/server/scripts/） */
+const REPO_ROOT = new URL('../../../', import.meta.url)
+
+function readSnapshot<T>(name: string): T {
+  return JSON.parse(readFileSync(new URL(name, REPO_ROOT), 'utf8')) as T
+}
+
+function asNumber(v: unknown, fallback: number): number {
+  return typeof v === 'number' && Number.isFinite(v) ? v : fallback
+}
+
+const db = getDb()
+const secrets = getKeySecrets()
+
+// ---------- 样例快照 ----------
+const logs = readSnapshot<{ data?: UpstreamLogRecord[] }>('api.log.token.json').data
+if (!Array.isArray(logs) || logs.length === 0) {
+  console.error('[seed] api.log.token.json 缺失或结构异常（缺 data[]）——请确认样例文件在仓库根')
+  process.exit(1)
+}
+const status = readSnapshot<{ data?: Record<string, unknown> }>('api.status.json').data ?? {}
+
+// ---------- 1. api_keys ----------
+const keyHash = secrets.keyHash(SEED_KEY)
+let [keyRow] = await db.select().from(apiKeys).where(eq(apiKeys.keyHash, keyHash)).limit(1)
+if (!keyRow) {
+  const first = logs[0]!
+  const other = parseOther(first.other)
+  const totalGranted = asNumber(other.subscription_total, 800_000_000)
+  const totalUsed = asNumber(other.subscription_used, 85_579_458)
+  const now = new Date()
+  ;[keyRow] = await db
+    .insert(apiKeys)
+    .values({
+      keyHash,
+      keyCiphertext: secrets.encryptKey(SEED_KEY),
+      status: 'active',
+      disabledSource: null,
+      consecutive401s: 0,
+      logSyncedAt: now,
+      usageSyncedAt: now,
+      tokenName: first.token_name ?? 'seed-token',
+      totalGranted,
+      totalUsed,
+      totalAvailable: totalGranted - totalUsed,
+      unlimitedQuota: false,
+      modelLimits: {},
+      modelLimitsEnabled: false,
+      expiresAt: null,
+    })
+    .returning()
+  console.log(`[seed] api_keys：已插入 id=${keyRow!.id}（token_name=${keyRow!.tokenName}）`)
+} else {
+  console.log(`[seed] api_keys：已存在（id=${keyRow.id}），跳过`)
+}
+
+// ---------- 2. usage_logs ----------
+const rows = logs.map((log) => toLogRow(keyRow!.id, log))
+const inserted = await db
+  .insert(usageLogs)
+  .values(rows)
+  .onConflictDoNothing({ target: usageLogs.dedupeKey })
+  .returning({ id: usageLogs.id })
+console.log(`[seed] usage_logs：新增 ${inserted.length} 条（样例共 ${logs.length} 条）`)
+
+// ---------- 3. qpu_history 基线 ----------
+const [qpuRow] = await db.select().from(qpuHistory).limit(1)
+if (!qpuRow) {
+  const qpu = asNumber(status.quota_per_unit, DEFAULT_QUOTA_PER_UNIT)
+  await db.insert(qpuHistory).values({ qpu, effectiveFrom: new Date() })
+  console.log(`[seed] qpu_history：已插入基线 qpu=${qpu}`)
+} else {
+  console.log(`[seed] qpu_history：已有数据（qpu=${qpuRow.qpu}），跳过`)
+}
+
+// ---------- 4. 示例 purchases（金额为占位值） ----------
+const [purchaseRow] = await db.select().from(purchases).limit(1)
+if (!purchaseRow) {
+  const first = logs[0]!
+  const other = parseOther(first.other)
+  const earliestSeconds = Math.min(...logs.map((l) => l.created_at))
+  const earliest = fromUnixSeconds(earliestSeconds)
+  await db.insert(purchases).values({
+    kind: 'subscription',
+    paidAt: earliest,
+    paidAmount: 168.5, // 占位值：实付金额只有作者知道
+    note: 'seed 示例（金额为占位值，非真实）',
+    periodStart: earliest,
+    periodEnd: null,
+    subscriptionId: asNumber(other.subscription_id, 90),
+    quotaGranted: asNumber(other.subscription_total, 800_000_000),
+  })
+  console.log('[seed] purchases：已插入示例一条（折算链验证用）')
+} else {
+  console.log(`[seed] purchases：已有数据（id=${purchaseRow.id}），跳过`)
+}
+
+console.log(`[seed] 完成。查询面可用假 key「${SEED_KEY}」命中已登记 key。`)

apps/server/src/lib/upstream-columns.ts

@@ -0,0 +1,19 @@
+/**
+ * 上游字段全组（`api_keys` 表 8 个上游列）——登记 / 启用 / 每轮额度采集共用一处，
+ * 避免三处各写一份导致漏列（api_keys 表设计 §3.4/§3.5：整组一起写）。
+ */
+import type { UpstreamTokenUsage } from '../upstream/types'
+import { expiresAtFromUnix } from './time'
+
+export function upstreamColumns(usage: UpstreamTokenUsage) {
+  return {
+    tokenName: usage.name,
+    totalGranted: usage.total_granted,
+    totalUsed: usage.total_used,
+    totalAvailable: usage.total_available,
+    unlimitedQuota: usage.unlimited_quota,
+    modelLimits: usage.model_limits,
+    modelLimitsEnabled: usage.model_limits_enabled,
+    expiresAt: expiresAtFromUnix(usage.expires_at),
+  }
+}

apps/server/src/routes/manage/api-keys.ts

@@ -5,7 +5,7 @@ import type { PatchApiKeyRequest, RegisterApiKeyRequest } from '@naku/shared'
 import { apiError, isUniqueViolation, upstreamErrorResponse } from '../../lib/api-error'
 import { getKeySecrets } from '../../lib/keys'
 import { toApiKeyRow } from '../../lib/serialize'
-import { expiresAtFromUnix } from '../../lib/time'
+import { upstreamColumns } from '../../lib/upstream-columns'
 import { manageAuth } from '../../middleware/manage-auth'
 import { getDb } from '../../db/client'
 import { apiKeys } from '../../db/schema'
@@ -21,20 +21,6 @@ export const apiKeysRoutes = new Hono()
 
 apiKeysRoutes.use(manageAuth)
 
-/** 上游字段全组（登记 / 启用验证成功时一起写，见 api_keys 表设计 §3.4/§3.5） */
-function upstreamColumns(usage: UpstreamTokenUsage) {
-  return {
-    tokenName: usage.name,
-    totalGranted: usage.total_granted,
-    totalUsed: usage.total_used,
-    totalAvailable: usage.total_available,
-    unlimitedQuota: usage.unlimited_quota,
-    modelLimits: usage.model_limits,
-    modelLimitsEnabled: usage.model_limits_enabled,
-    expiresAt: expiresAtFromUnix(usage.expires_at),
-  }
-}
-
 /** 清单（按 id 升序，不分页） */
 apiKeysRoutes.get('/', async (c) => {
   const rows = await getDb().select().from(apiKeys).orderBy(asc(apiKeys.id))

apps/server/src/sync/dedupe.test.ts

@@ -0,0 +1,53 @@
+/**
+ * 去重键单测（usage_logs 表设计 §3.1 + 实施计划「实现假设」#5）。
+ */
+import { describe, expect, test } from 'bun:test'
+
+import { contentFingerprint, dedupeKeyFor } from './dedupe'
+import type { UpstreamLogRecord } from '../upstream/types'
+
+const log = (over: Partial<UpstreamLogRecord> = {}): UpstreamLogRecord => ({
+  id: 1,
+  user_id: 5,
+  created_at: 1789126642,
+  type: 2,
+  content: '',
+  username: 'chendong',
+  token_name: 'yangsiyue',
+  model_name: 'glm-5.3-flash',
+  quota: 702,
+  prompt_tokens: 29157,
+  completion_tokens: 1036,
+  use_time: 30,
+  is_stream: true,
+  channel: 7,
+  channel_name: '',
+  token_id: 29,
+  group: 'default',
+  ip: '',
+  request_id: '202609111136522416558528268d9d6CRxC30kh',
+  other: '{}',
+  ...over,
+})
+
+describe('dedupeKeyFor', () => {
+  test('request_id 存在 → `${key_id}:${request_id}`', () => {
+    expect(dedupeKeyFor(3, log())).toBe('3:202609111136522416558528268d9d6CRxC30kh')
+  })
+
+  test('跨 key 隔离：同一 request_id 在不同 key 下不同键', () => {
+    expect(dedupeKeyFor(3, log())).not.toBe(dedupeKeyFor(4, log()))
+  })
+
+  test('request_id 为空 → 内容指纹（同输入稳定）、前缀区分 `fp:`', () => {
+    const a = dedupeKeyFor(3, log({ request_id: '' }))
+    const b = dedupeKeyFor(3, log({ request_id: '' }))
+    expect(a).toBe(b)
+    expect(a.startsWith('fp:')).toBe(true)
+  })
+
+  test('指纹随内容变化（真实不同请求不同键）', () => {
+    expect(contentFingerprint(3, log())).not.toBe(contentFingerprint(3, log({ quota: 703 })))
+    expect(contentFingerprint(3, log())).not.toBe(contentFingerprint(4, log()))
+  })
+})

apps/server/src/sync/dedupe.ts

@@ -0,0 +1,26 @@
+/**
+ * 去重键生成（usage_logs 表设计 §3.1 + 实施计划「实现假设」#5）：
+ * `dedupe_key = ${key_id}:${request_id}`；`request_id` 为空时退化为内容指纹
+ * `sha256(key_id｜created_at｜model_name｜quota｜prompt_tokens｜completion_tokens)`。
+ */
+import { createHash } from 'node:crypto'
+
+import type { UpstreamLogRecord } from '../upstream/types'
+
+/** 内容指纹（仅 `request_id` 缺失时启用） */
+export function contentFingerprint(keyId: number, log: UpstreamLogRecord): string {
+  const parts = [
+    keyId,
+    log.created_at,
+    log.model_name ?? '',
+    log.quota,
+    log.prompt_tokens,
+    log.completion_tokens,
+  ].join('|')
+  return createHash('sha256').update(parts, 'utf8').digest('hex')
+}
+
+export function dedupeKeyFor(keyId: number, log: UpstreamLogRecord): string {
+  const requestId = (log.request_id ?? '').trim()
+  return requestId ? `${keyId}:${requestId}` : `fp:${contentFingerprint(keyId, log)}`
+}

apps/server/src/sync/index.ts

@@ -1,24 +1,206 @@
 /**
- * 同步引擎（见 `docs/design/同步引擎设计.md`）——骨架占位。
+ * 同步引擎（见 `docs/design/同步引擎设计.md`）：调度 + 单轮执行。
  *
- * 已定骨架（§3.2 / §3.3 / §3.4）：
- * - 后端进程内定时器（不引 worker、不拆进程）
- * - 周期 5 分钟，固定 tick + 防重入
- * - 启动立即跑一轮
- * - 单轮串行遍历全部 active key
- * - 每轮：日志（整批单事务入库 → log_synced_at 取请求发起时刻）· 额度（上游字段全组 + usage_synced_at）· QPU 采样（每轮一次、值变化才追加）
- * - 失败：401 计数（N=3 摘除）· 403/500/网络不计数 · 不单轮重试
+ * 结构（§3.2 / §3.3 / §3.4）：
+ * - 后端进程内定时器（不引 worker、不拆进程）；周期 5 分钟，固定 tick + 防重入；启动立即跑一轮
+ * - 单轮串行遍历全部 `active` key；每 key 依次：日志（整批单事务入库 → `log_synced_at` 取请求发起时刻）
+ *   → 额度（上游字段全组 + `usage_synced_at`，与日志各自独立 try）
+ * - 每轮一次 QPU 采样（独立于 key 循环；`qpu_history` 为空时插基线）
+ * - 失败：401 计数（N=3 摘除，计数只认 401）· 403/500/网络不计数 · 不单轮重试
+ * - 可观测：`[sync]` 前缀一行式日志；每轮摘要；`log_synced_at` 落后 > 24h 每轮 warning
  */
+import { asc, desc, eq } from 'drizzle-orm'
 
-const TICK_MS = 5 * 60 * 1000
+import { getDb } from '../db/client'
+import { apiKeys, qpuHistory, usageLogs } from '../db/schema'
+import { getKeySecrets } from '../lib/keys'
+import { upstreamColumns } from '../lib/upstream-columns'
+import { getUpstreamClient } from '../upstream/client'
+import { errorMessage, UpstreamError } from '../upstream/errors'
+import type { UpstreamLogRecord } from '../upstream/types'
+import { toLogRow } from './log-row'
+import {
+  combineRoundResult,
+  nextKeyCounter,
+  STALE_WARN_MS,
+  TICK_MS,
+  type CallResult,
+} from './policy'
+
+type KeyRow = typeof apiKeys.$inferSelect
+
+interface RoundSummary {
+  keyCount: number
+  okCount: number
+  failCount: number
+  inserted: number
+}
 
 let running = false
 
-/** 一轮：遍历全部 active key（串行）→ 日志 / 额度；每轮一次 QPU 采样 */
-async function runOnce(): Promise<void> {
-  // TODO(#5): 日志同步 —— GET /api/log/token → 整批单事务入库（ON CONFLICT (dedupe_key) DO NOTHING）→ 写 log_synced_at
-  // TODO(#6): 额度采集 —— GET /api/usage/token → 上游字段全组 + usage_synced_at（与日志各自独立）
-  // TODO(#7): QPU 采样 —— GET /api/status → 值变化才追加 qpu_history（每轮一次，独立于 key 循环）
+/** 上游错误 → 本轮结果：只有 401 是「key 失效」信号（403 = 账号封禁，不计数） */
+function classify(err: unknown): CallResult {
+  if (err instanceof UpstreamError && err.kind === 'key_rejected' && err.status === 401) {
+    return 'unauthorized'
+  }
+  return 'failed'
+}
+
+/**
+ * 日志整批单事务入库（`ON CONFLICT (dedupe_key) DO NOTHING`）+ 写 `log_synced_at`
+ * ——全成或全不写：失败则「数据截止」不前移，下一轮幂等重来（§3.3）。
+ */
+async function persistLogs(
+  keyId: number,
+  logs: UpstreamLogRecord[],
+  syncedAt: Date,
+): Promise<number> {
+  const db = getDb()
+  const rows = logs.map((log) => toLogRow(keyId, log))
+  return await db.transaction(async (tx) => {
+    let inserted = 0
+    if (rows.length > 0) {
+      const res = await tx
+        .insert(usageLogs)
+        .values(rows)
+        .onConflictDoNothing({ target: usageLogs.dedupeKey })
+        .returning({ id: usageLogs.id })
+      inserted = res.length
+    }
+    await tx.update(apiKeys).set({ logSyncedAt: syncedAt }).where(eq(apiKeys.id, keyId))
+    return inserted
+  })
+}
+
+/** 单 key 一轮：日志 → 额度（各自独立 try）→ 状态机（每轮一次） */
+async function syncOneKey(key: KeyRow): Promise<{ round: CallResult; inserted: number }> {
+  const db = getDb()
+  let plainKey: string
+  try {
+    plainKey = getKeySecrets().decryptKey(Buffer.from(key.keyCiphertext))
+  } catch (err) {
+    // 解密失败：记日志、跳过、不影响其他 key；不计数、不摘除（§3.4）
+    console.error(
+      `[sync] key_id=${key.id} 凭证解密失败，跳过本轮（需人工重新登记）：${errorMessage(err)}`,
+    )
+    return { round: 'failed', inserted: 0 }
+  }
+
+  const results: CallResult[] = []
+  let inserted = 0
+
+  // —— 日志同步（独立 try）——
+  const logRequestedAt = new Date()
+  const logStartedMs = Date.now()
+  try {
+    const logs = await getUpstreamClient().fetchLogs(plainKey)
+    inserted = await persistLogs(key.id, logs, logRequestedAt)
+    results.push('ok')
+    console.log(
+      `[sync] key_id=${key.id} 日志同步：上游 ${logs.length} 条，新增 ${inserted} 条（${Date.now() - logStartedMs}ms）`,
+    )
+  } catch (err) {
+    results.push(classify(err))
+    console.error(
+      `[sync] key_id=${key.id} 日志同步失败（${Date.now() - logStartedMs}ms）：${errorMessage(err)}`,
+    )
+  }
+
+  // —— 额度采集（独立 try；与日志互不影响、各自写各自的时间戳）——
+  const usageRequestedAt = new Date()
+  const usageStartedMs = Date.now()
+  try {
+    const usage = await getUpstreamClient().fetchTokenUsage(plainKey)
+    await db
+      .update(apiKeys)
+      .set({ usageSyncedAt: usageRequestedAt, ...upstreamColumns(usage) })
+      .where(eq(apiKeys.id, key.id))
+    results.push('ok')
+  } catch (err) {
+    results.push(classify(err))
+    console.error(
+      `[sync] key_id=${key.id} 额度采集失败（${Date.now() - usageStartedMs}ms）：${errorMessage(err)}`,
+    )
+  }
+
+  // —— 401 状态机（每轮一次；401 优先于成功——见实施计划「实现假设」#15）——
+  const round = combineRoundResult(results)
+  const next = nextKeyCounter(
+    {
+      consecutive401s: key.consecutive401s,
+      status: key.status as 'active' | 'disabled',
+      disabledSource: key.disabledSource as 'manual' | 'auto_401' | null,
+    },
+    round,
+  )
+  const patch: Partial<typeof apiKeys.$inferInsert> = {}
+  if (next.consecutive401s !== key.consecutive401s) patch.consecutive401s = next.consecutive401s
+  if (next.status !== key.status) patch.status = next.status
+  if (next.disabledSource !== key.disabledSource) patch.disabledSource = next.disabledSource
+  if (Object.keys(patch).length > 0) {
+    await db.update(apiKeys).set(patch).where(eq(apiKeys.id, key.id))
+  }
+  if (next.removed) {
+    console.warn(
+      `[sync] key_id=${key.id} 连续 ${next.consecutive401s} 次 401 → 自动摘除（disabled_source=auto_401）`,
+    )
+  }
+
+  return { round, inserted }
+}
+
+/** QPU 采样：每轮一次；`qpu_history` 为空时插基线，值变化才追加（§3.3 / §3.6） */
+async function sampleQpu(): Promise<void> {
+  const db = getDb()
+  try {
+    const status = await getUpstreamClient().fetchStatus()
+    const qpu = status.quota_per_unit
+    const [last] = await db
+      .select()
+      .from(qpuHistory)
+      .orderBy(desc(qpuHistory.effectiveFrom))
+      .limit(1)
+    const effectiveFrom = new Date()
+    if (!last) {
+      await db.insert(qpuHistory).values({ qpu, effectiveFrom })
+      console.log(`[sync] QPU 基线已插入：${qpu}`)
+    } else if (Number(last.qpu) !== qpu) {
+      await db.insert(qpuHistory).values({ qpu, effectiveFrom })
+      console.log(`[sync] QPU 变化：${last.qpu} → ${qpu}`)
+    }
+  } catch (err) {
+    console.error(`[sync] QPU 采样失败：${errorMessage(err)}`)
+  }
+}
+
+/** 一轮：遍历全部 active key（串行）→ 日志 / 额度；随后 QPU 采样（无 key 也跑） */
+async function runOnce(): Promise<RoundSummary> {
+  const db = getDb()
+  const keys = await db
+    .select()
+    .from(apiKeys)
+    .where(eq(apiKeys.status, 'active'))
+    .orderBy(asc(apiKeys.id))
+
+  const summary: RoundSummary = { keyCount: keys.length, okCount: 0, failCount: 0, inserted: 0 }
+  const nowMs = Date.now()
+
+  for (const key of keys) {
+    // 「3 天窗口」警示：落后 > 24h 每轮 warning（§3.5）
+    if (key.logSyncedAt && nowMs - key.logSyncedAt.getTime() > STALE_WARN_MS) {
+      const hours = Math.floor((nowMs - key.logSyncedAt.getTime()) / 3_600_000)
+      console.warn(
+        `[sync] key_id=${key.id} 数据截止已落后 ${hours} 小时（>24h；窗口上限约 3 天，请尽快恢复同步）`,
+      )
+    }
+    const { round, inserted } = await syncOneKey(key)
+    summary.inserted += inserted
+    if (round === 'ok') summary.okCount++
+    else summary.failCount++
+  }
+
+  await sampleQpu()
+  return summary
 }
 
 async function tick(): Promise<void> {
@@ -27,11 +209,14 @@ async function tick(): Promise<void> {
     return
   }
   running = true
+  const startedMs = Date.now()
   try {
-    await runOnce()
-    console.log('[sync] 轮次完成（骨架占位：暂无实际采集动作）')
+    const s = await runOnce()
+    console.log(
+      `[sync] 轮次完成：${s.keyCount} key，成功 ${s.okCount}，失败 ${s.failCount}，新增日志 ${s.inserted} 条，耗时 ${((Date.now() - startedMs) / 1000).toFixed(1)} 秒`,
+    )
   } catch (err) {
-    console.error('[sync] 轮次异常', err)
+    console.error(`[sync] 轮次异常：${errorMessage(err)}`)
   } finally {
     running = false
   }

apps/server/src/sync/log-row.ts

@@ -0,0 +1,45 @@
+/**
+ * 上游日志行 → `usage_logs` 列值（同步引擎与 seed 脚本共用，避免两份映射漂移）。
+ */
+import { fromUnixSeconds } from '../lib/time'
+import type { usageLogs } from '../db/schema'
+import type { UpstreamLogRecord } from '../upstream/types'
+import { dedupeKeyFor } from './dedupe'
+
+/** 上游 `other`（JSON 字符串）→ jsonb；解析失败退化为空对象（保真优先，不丢整条日志） */
+export function parseOther(raw: string | undefined | null): Record<string, unknown> {
+  if (!raw) return {}
+  try {
+    const v = JSON.parse(raw) as unknown
+    return v && typeof v === 'object' && !Array.isArray(v) ? (v as Record<string, unknown>) : {}
+  } catch {
+    return {}
+  }
+}
+
+export function toLogRow(keyId: number, log: UpstreamLogRecord): typeof usageLogs.$inferInsert {
+  return {
+    keyId,
+    dedupeKey: dedupeKeyFor(keyId, log),
+    userId: log.user_id,
+    createdAt: fromUnixSeconds(log.created_at),
+    type: log.type,
+    content: log.content ?? '',
+    username: log.username ?? '',
+    tokenName: log.token_name ?? '',
+    modelName: log.model_name ?? '',
+    quota: log.quota,
+    promptTokens: log.prompt_tokens,
+    completionTokens: log.completion_tokens,
+    useTime: log.use_time,
+    isStream: log.is_stream,
+    channel: log.channel,
+    channelName: log.channel_name ?? '',
+    tokenId: log.token_id,
+    groupName: log.group ?? 'default',
+    ip: log.ip ?? '',
+    requestId: log.request_id ?? '',
+    upstreamRequestId: log.upstream_request_id ?? null,
+    other: parseOther(log.other),
+  }
+}

apps/server/src/sync/policy.test.ts

@@ -0,0 +1,79 @@
+/**
+ * 401 状态机与轮次结果合并的单测（同步引擎设计 §3.4 + 实施计划「实现假设」#15）。
+ */
+import { describe, expect, test } from 'bun:test'
+
+import {
+  combineRoundResult,
+  MAX_CONSECUTIVE_401S,
+  nextKeyCounter,
+  type KeyCounterNext,
+} from './policy'
+
+const active = (consecutive401s = 0): KeyCounterNext => ({
+  consecutive401s,
+  status: 'active',
+  disabledSource: null,
+  removed: false,
+})
+
+describe('combineRoundResult（两个调用合并为「本轮结果」）', () => {
+  test('任一 401 → unauthorized（401 优先于成功）', () => {
+    expect(combineRoundResult(['ok', 'unauthorized'])).toBe('unauthorized')
+    expect(combineRoundResult(['unauthorized', 'failed'])).toBe('unauthorized')
+  })
+
+  test('无 401 且任一成功 → ok', () => {
+    expect(combineRoundResult(['ok', 'failed'])).toBe('ok')
+    expect(combineRoundResult(['ok', 'ok'])).toBe('ok')
+  })
+
+  test('全失败 → failed（不动计数）', () => {
+    expect(combineRoundResult(['failed', 'failed'])).toBe('failed')
+  })
+})
+
+describe('nextKeyCounter（401 计数 / 摘除 / 清零）', () => {
+  test('成功 → 计数清零（不摘除）', () => {
+    const next = nextKeyCounter(active(2), 'ok')
+    expect(next.consecutive401s).toBe(0)
+    expect(next.status).toBe('active')
+    expect(next.removed).toBe(false)
+  })
+
+  test('其他失败（403/500/网络）→ 计数不动', () => {
+    const next = nextKeyCounter(active(1), 'failed')
+    expect(next.consecutive401s).toBe(1)
+    expect(next.removed).toBe(false)
+  })
+
+  test('连续 401 累加；未达阈值不摘除', () => {
+    let state = active(0)
+    for (let i = 1; i < MAX_CONSECUTIVE_401S; i++) {
+      const next = nextKeyCounter(state, 'unauthorized')
+      expect(next.consecutive401s).toBe(i)
+      expect(next.removed).toBe(false)
+      expect(next.status).toBe('active')
+      state = next
+    }
+  })
+
+  test('第 N 次（N=3）连续 401 → 摘除且 disabled_source = auto_401', () => {
+    let state: KeyCounterNext = active(0)
+    for (let i = 1; i <= MAX_CONSECUTIVE_401S; i++) {
+      state = nextKeyCounter(state, 'unauthorized')
+    }
+    expect(state.consecutive401s).toBe(MAX_CONSECUTIVE_401S)
+    expect(state.status).toBe('disabled')
+    expect(state.disabledSource).toBe('auto_401')
+    expect(state.removed).toBe(true)
+  })
+
+  test('中途成功会打断连续性（计数回零）', () => {
+    let state: KeyCounterNext = active(2)
+    state = nextKeyCounter(state, 'ok')
+    state = nextKeyCounter(state, 'unauthorized')
+    expect(state.consecutive401s).toBe(1)
+    expect(state.removed).toBe(false)
+  })
+})

apps/server/src/sync/policy.ts

@@ -0,0 +1,60 @@
+/**
+ * 同步策略常量与纯决策函数（同步引擎设计 §3.2/§3.4/§3.5）。
+ * 抽成纯函数是为了把「401 计数 / 摘除」与「本轮结果合并」的规则钉在单测里。
+ */
+
+/** 一轮的固定周期：5 分钟（§3.2） */
+export const TICK_MS = 5 * 60 * 1000
+
+/** 连续 401 摘除阈值 N = 3（§3.4） */
+export const MAX_CONSECUTIVE_401S = 3
+
+/** 落后警示阈值：`log_synced_at` 落后超过 24 小时 → 每轮 warning（§3.5，距 3 天窗口留 2 天余量） */
+export const STALE_WARN_MS = 24 * 60 * 60 * 1000
+
+/** 单个上游调用在本轮的结果 */
+export type CallResult = 'ok' | 'unauthorized' | 'failed'
+
+/**
+ * 一轮内两个调用（日志 / 额度）的合并判定（实施计划「实现假设」#15）：
+ * 任一 401 → `unauthorized`（key 失效信号优先）；否则任一成功 → `ok`；全失败 → `failed`（不动计数）。
+ */
+export function combineRoundResult(results: CallResult[]): CallResult {
+  if (results.includes('unauthorized')) return 'unauthorized'
+  if (results.includes('ok')) return 'ok'
+  return 'failed'
+}
+
+export interface KeyCounterState {
+  consecutive401s: number
+  status: 'active' | 'disabled'
+  disabledSource: 'manual' | 'auto_401' | null
+}
+
+export interface KeyCounterNext extends KeyCounterState {
+  /** 本轮是否触发自动摘除 */
+  removed: boolean
+}
+
+/**
+ * 401 状态机（§3.4）：401 → 计数 +1（达 N 摘除，`disabled_source = auto_401`）；
+ * 成功 → 计数清零；其他失败（403 / 500 / 网络 / 解密失败）→ 计数不动。
+ */
+export function nextKeyCounter(state: KeyCounterState, round: CallResult): KeyCounterNext {
+  if (round === 'ok') {
+    return { ...state, consecutive401s: 0, removed: false }
+  }
+  if (round === 'failed') {
+    return { ...state, removed: false }
+  }
+  const consecutive401s = state.consecutive401s + 1
+  if (consecutive401s >= MAX_CONSECUTIVE_401S) {
+    return {
+      consecutive401s,
+      status: 'disabled',
+      disabledSource: 'auto_401',
+      removed: true,
+    }
+  }
+  return { ...state, consecutive401s, removed: false }
+}

docs/design/实施计划.md

@@ -7,7 +7,7 @@
 | 状态 | 定稿（2026-09-14；**进度表随 loop 持续更新**） |
 | 关联 | `docs/design/API契约.md` · `docs/design/同步引擎设计.md` · `docs/design/交互设计.md` · 四张表字段设计 · `AGENTS.md`（命令） |
 | 范围 | 阶段划分、每批验收标准、loop 机制与推进规则；技术细节一律以设计文档为准 |
-| 当前进度 | 阶段 0–3 已完成；下一批：**阶段 4**（同步引擎：tick / 日志入库 / 额度 / QPU / 401 状态机 / seed） |
+| 当前进度 | 阶段 0–4 已完成；下一批：**阶段 5**（查看面数据面：`logs/query` + `summary` + 折算链） |
 
 ---
 
@@ -40,6 +40,8 @@
 | 12 | 登记 / 启用验证成功即写 `usage_synced_at` + 上游字段全组（该次 `GET /api/usage/token` 本身即一次额度采集）；`log_synced_at` 仍只由日志同步写入 | 契约 §2.2 已定登记响应 `log_synced_at: null`；额度侧取「数据已入库就不撒谎」口径 |
 | 13 | 失败分类：401 / 403 → `key_rejected`（上游明确拒绝）；其余非 2xx / 网络故障 / 超时 / 响应结构异常 → `upstream_unavailable` | 上游事实：403 = 账号封禁；可调 |
 | 14 | `PATCH` 幂等分支不重复调上游验证（直接返回当前行）；启用时凭证解密失败 → `500 internal`（提示需重新登记） | 与「解密失败不摘除、需人工重新登记」一致；可调 |
+| 15 | 一轮内「日志 + 额度」两个调用的合并判定：**任一 401 → 本轮计 1 次 401**（401 优先）；否则任一成功 → 计数清零；全失败 → 计数不动 | 设计（§3.4）未覆盖「一成功一 401」的冲突情形；401 优先 + N=3 阈值使单次误报无害 |
+| 16 | seed 脚本用假 key 常量 `sk-seed-demo-key-0001`（`SEED_KEY` 可覆盖）；`apps/server/scripts/seed.ts` 幂等可重跑 | 实施计划 §6 已定「假 key 常量」；仅开发期 |
 
 ## 2. 前提（2026-09-14 定）
 
@@ -55,7 +57,7 @@
 | 1 | **数据层与密钥**：crypto（HKDF 派生 / HMAC-SHA256 / AES-256-GCM）· 时间工具（+08:00 序列化、UTC 秒转换）· 建表 | 单测：哈希确定性 + 加解密往返；`db:push` 后四张表在库 | ✅ 2026-09-14（15 项单测绿；四表在库——`api_keys` 17 列 / `usage_logs` 23 列 / 索引 8 个） |
 | 2 | **上游客户端 + key 管理**（`GET` / `POST` / `PATCH` 三端点）· 客户端接口化（§6） | curl 实测：登记 → 清单 → 停用/启用（**注入样例实现**：登记成功路径用样例响应驱动）；三类错误码分支正确（`key_rejected` / `already_registered` / `upstream_unavailable`） | ✅ 2026-09-15（28 项单测绿；curl 16 步——登记 201 / 重复 409 / `key_rejected` 400 / `upstream_unavailable` 502 / 停用与幂等 / 启用清 `consecutive_401s` / `not_found` 404 / 非法参数 400；顺带修正「`load-env` 未被服务端接线」） |
 | 3 | **管理面其余**：purchases 五端点 + qpu 一端点 | curl 实测：CRUD 全流程 + 观测聚合（`matched_purchase_id` 判定正确） | ✅ 2026-09-15（33 项单测绿；curl 16 步——登记 201 / 缺必填与非法值与期间顺序 400 / 部分更新与显式清空 / 排序 / 404 / 物理删除 / 观测聚合 `matched_purchase_id` 判定正确） |
-| 4 | **同步引擎**：tick（5 分钟 + 防重入 + 启动即跑）· 日志整批事务入库 · 额度采集 · QPU 采样 · 401 状态机（N=3）· **seed 脚本**（§6） | 库中出现数据（seed 或真实采集）；`[sync]` 日志符合引擎设计；401 状态机单测（连续 401 → 摘除且 `disabled_source = auto_401`） | ☐ |
+| 4 | **同步引擎**：tick（5 分钟 + 防重入 + 启动即跑）· 日志整批事务入库 · 额度采集 · QPU 采样 · 401 状态机（N=3）· **seed 脚本**（§6） | 库中出现数据（seed 或真实采集）；`[sync]` 日志符合引擎设计；401 状态机单测（连续 401 → 摘除且 `disabled_source = auto_401`） | ✅ 2026-09-15（45 项单测绿；seed 灌入 984 条日志 + QPU 基线 + 示例购买；引擎实测——幂等重跑新增 0 条、摘要日志符合设计、**401 端到端：1→2→3 次摘除且摘除后不再轮询**） |
 | 5 | **查看面数据面**：`logs/query` + `summary` + 折算链（按购买期间分段取尺） | curl 实测：字段齐全、`meta.data_until` 取最慢、金额按购买分段累加；**无归属购买的行 `amount_rmb = null` 且计入 `unpriced_count`**（2026-09-14 定） | ☐ |
 | 6 | **前端 `/usage-log`**：筛选区（key 标签 + datetimerange + 快捷项）· 卡片 · 表格 + 展开行 · 分页器 · 六种状态模式 | 浏览器实测：seed 数据走一遍；六态逐一可见（加载 / 初始 / 未收录 / 无记录 / 同步中 / 失败）；金额为 `null` 显示「—」 | ☐ |
 | 7 | **前端管理面**：口令门帘 + 侧边栏 + 三页（key 管理 / 购买记录含核对提示与补录 / 换算参数） | 浏览器实测：登记、启停、CRUD、补录、时间线全操作 | ☐ |
@@ -69,6 +71,7 @@
 | 2 | 充值（`wallet`）路径的折算近似（设计文档标「后议」） | 先按「未折算」处理（`null` / 计入 `unpriced_count`）；池子加权平均待后议 |
 | 3 | `dedupe_key` 内容指纹的拼接细节 | **已决**：见「实现假设」#5（表设计 §3.1 已给字段清单） |
 | 4 | 上游对同账号请求的态度（限流 / 并发被拒时的表现） | 遇错即按「其他故障」处理（不计数、不摘除），如实记录日志 |
+| 5 | **`request_id` 在错误行上不唯一**（2026-09-15 实测样板：8 个 `request_id` 各对应 3 条 `type=5` 错误记录，共 24 行）——当前按表设计原样实现 `dedupe_key = key_id:request_id`，同 `request_id` 的多条错误行**合并为 1 条** | **待作者裁决**：影响 = 失败请求次数（`error_count`）少算；候选方案 = 错误行去重键后缀内容指纹 / 序号（`usage_logs` 表设计 §3.1 需相应修订） |
 
 ## 5. 风险
 

docs/upstream/NewAPI-sk-key接口文档.md

@@ -196,7 +196,7 @@ Authorization: Bearer sk-xxxx
 | `token_id` | int | 上游 token ID（对账用） |
 | `group` | string | 分组名（默认 `default`） |
 | `ip` | string | 请求 IP（普通用户视角为空串） |
-| `request_id` | string | 请求 ID（格式：`yyyyMMddHHmmss` 前缀 + 随机串）；JSON tag 为 `omitempty`，但源码写入时有兜底保证非空——**去重唯一依据** |
+| `request_id` | string | 请求 ID（格式：`yyyyMMddHHmmss` 前缀 + 随机串）；JSON tag 为 `omitempty`，但源码写入时有兜底保证非空——**去重唯一依据**。**补充实测（2026-09-15）**：错误场景下 `request_id` **不保证行级唯一**——1000 条实测样本中有 8 个 `request_id` 各对应 3 条 `type=5` 错误记录（同一请求的多次失败 / 重试，`content` 的 `status_code` 可不同）；按 `request_id` 去重会把它们合并为 1 条（消费类 `type=2` 样本中未观察到重复） |
 | `upstream_request_id` | string | 上游请求 ID；JSON tag 为 `omitempty`——**空值时不出现、有值时才出现**（实测样本中缺席）；消费方需按「可选字段」处理 |
 | `other` | string | JSON 字符串，扩展信息（结构见下） |
 

package.json

@@ -17,7 +17,8 @@
     "type-check": "bun run --filter '@naku/web' type-check && bunx tsc --noEmit -p apps/server",
     "db:generate": "bun run --filter '@naku/server' db:generate",
     "db:migrate": "bun run --filter '@naku/server' db:migrate",
-    "db:push": "bun run --filter '@naku/server' db:push"
+    "db:push": "bun run --filter '@naku/server' db:push",
+    "seed": "bun run --filter '@naku/server' seed"
   },
   "devDependencies": {
     "typescript": "^6.0.0"