refactor(web): API 路由统一 /api 前缀 + 完善 OpenAPI 文档

- 路由挂 APIRouter(prefix="/api")，健康检查路径 /health -> /api/health，后期加接口（/api/compare、/api/queue 等）自动共享前缀。 - 文档路径归到 /api 下：/api/docs、/api/redoc、/api/openapi.json （原 /docs、/redoc、/openapi.json 已 404）。 - 新增 Pydantic 响应模型 schemas.py，并在 /api/health 挂 response_model，使 /api/docs 展示完整响应结构。 - 为接口和每个响应字段补详细英文 description（覆盖范围、status 三态判定、 HTTP 码含义、各字段语义如 cycle_lag 阈值/dead 含义等）。不动业务逻辑与同步主循环，仅 Web 层。
feat(health): 被动式健康检查 API（FastAPI，覆盖服务健康+数据健康）
2026-08-05 12:55:45 +08:00 · 2026-08-05 12:13:14 +08:00 · 2026-08-05 11:00:58 +08:00
19 changed files with 2073 additions and 68 deletions
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -21,6 +21,12 @@ logging:
  level: INFO
  path: "<LOG_PATH>"
 health:
  enabled: true           # 开启被动式健康检查 HTTP endpoint
  host: "0.0.0.0"         # 监听地址，0.0.0.0 内网/FRP 均可访问
  port: 8421              # 健康检查端口
  log_level: "warning"    # uvicorn 自身日志级别，避免刷屏
 files:
  - {file: "一车间.accdb", root: 2026, schema: "workshopOne", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "一车间每日催货落实记录_停"]}
  - {file: "二车间.accdb", root: 2026, schema: "workshopTwo", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
--- a/docs/incremental-sync-flow.md
+++ b/docs/incremental-sync-flow.md
@@ -0,0 +1,281 @@
 # 增量同步流程详解（Incremental Sync Flow）
 > 本文档梳理 Access → SQL Server 增量同步的完整流程，逐节点说明「发生了什么、读了/写了什么、状态如何流转」，作为重构「Insert 降级 Delete」逻辑的决策依据。
 >
 > 涉及代码：`src/sync/service.py`（主循环 cycle）、`src/sync/capture.py`（捕获）、`sql/02_sync_apply.sql`（应用）、`src/sync/cleanup.py`（清理）、`src/sync/sql_writer.py`（SQL 端读写）、`src/sync/access_reader.py`（Access 端读取）。
 ---
 ## 一、整体架构：一轮 cycle 的三段流水线
 每个 cycle（默认间隔 `poll_interval_seconds=10s`）跑一遍三段，顺序固定、不可调换：
 ```mermaid
 flowchart TD
    START([cycle 开始<br/>分配 cycle_id]) --> CAP
    CAP["【1. Capture 捕获】<br/>逐库读 TableChangeLog → 回读整行 → 入队 SyncQueue<br/><i>src/sync/capture.py</i>"]
    CAP --> APPLY
    APPLY["【2. Apply 应用】<br/>调 usp_SyncApply 把 SyncQueue pending 行落到镜像表<br/><i>sql/02_sync_apply.sql</i>"]
    APPLY --> HEALTH
    HEALTH["【2.5 队列健康检查】<br/>查 error/dead 卡死行并告警"]
    HEALTH --> CLEAN
    CLEAN["【3. Cleanup 清理】<br/>删 Access 已应用日志 → SyncQueue 行标 cleaned<br/><i>src/sync/cleanup.py</i>"]
    CLEAN --> PURGE
    PURGE["【3.5 Purge 回收】<br/>删 SyncQueue 中超保留期的 cleaned 行"]
    PURGE --> DONE([cycle 结束<br/>休眠 poll_interval])
    style CAP fill:#e3f2fd,stroke:#1976d2
    style APPLY fill:#fff3e0,stroke:#f57c00
    style HEALTH fill:#fce4ec,stroke:#c2185b
    style CLEAN fill:#e8f5e9,stroke:#388e3c
    style PURGE fill:#f3e5f5,stroke:#7b1fa2
 ```
 **关键设计约束**（决定重构可行性的红线）：
 - **Access 的 `TableChangeLog` 是 append-only，无状态字段**——它只是个待处理队列，无法在上面记录「已重试几次」。
 - **日志清除的唯一依据是 SyncQueue 的 `Status='applied'`**——只要一条日志对应的队列行不是 applied，cleanup 就不会删它（详见第三节）。
 - **SyncQueue 有唯一索引 `(SourceFile,SourceTable,SourceLogID)` 去重**——同一日志第二次入队会被静默跳过，不会覆盖原行、不会自增计数。
 ---
 ## 二、Capture 阶段（数据捕获）—— ❗重构的核心战场
 逐库处理，每个 Access 文件独立隔离（单库失败不影响其它）。
 ```mermaid
 flowchart TD
    A([开始 capture 某个文件]) --> B["读 Access TableChangeLog<br/>最旧 N 条（按 ID 升序）<br/>N = capture_batch_size=500"]
    B --> C{有日志行?}
    C -- 否 --> Z([capture 结束])
    C -- 是 --> D[逐行处理]
    D --> E{"表是否在同步范围内?<br/>is_synced_table"}
    E -- 否 --> F["跳过（out_of_scope）<br/>不计入，下轮仍会读到"]
    F --> D
    E -- 是 --> G{"OperateType?"}
    G -- Insert/Update --> H["🔑 回读整行<br/>read_row(table, record_id)<br/>SELECT * FROM 表 WHERE ID=?"]
    H --> I{读到行?}
    I -- 是 --> J["row_data = JSON 序列化<br/>op 保持 Insert/Update"]
    I -- ❌否 --> K["⚠️ 降级 op = Delete<br/>row_data = None<br/>写 DOWNGRADE 警告日志"]
    G -- Delete --> L["op = Delete<br/>row_data = None<br/>（不回读，Delete 无需数据）"]
    G -- 其它未知 --> M["跳过（unknown_op）<br/>下轮仍会读到"]
    J --> N["写 SyncLogArchive（永久审计）<br/>记录 OriginalOperateType + ProcessedOperateType"]
    K --> N
    L --> N
    N --> O["入队 SyncQueue（去重插入）<br/>INSERT...WHERE NOT EXISTS"]
    O --> P{插入成功?}
    P -- 是 --> Q["enqueued +1"]
    P -- 否（去重命中）--> R["dedup_skipped +1<br/>说明上轮 apply 失败残留"]
    Q --> S{还有下一行?}
    R --> S
    F --> S
    M --> S
    S -- 是 --> D
    S -- 否 --> T["打印 capture summary<br/>read/enqueued/downgraded/..."]
    T --> Z
    style K fill:#ffcdd2,stroke:#c62828,stroke-width:3px
    style H fill:#fff9c4,stroke:#f9a825
    style I fill:#fff9c4,stroke:#f9a825
 ```
 ### 🔴 问题节点：降级 Delete（`capture.py:93-108`）
 ```python
 if op in ("Insert", "Update"):
    d = reader.read_row(lr.table_name, lr.record_id)
    if d is None:
        op = "Delete"          # ← 问题根源：读不到就降级
        st.downgraded += 1
        log.warning("capture DOWNGRADE %s->Delete ...")
 ```
 **为什么读不到？两个场景无法区分：**
 1. **真删除**：行被 Insert 后又 Delete（Access 客户端先插后删）→ 这时降级 Delete 是「碰巧正确」。
 2. **可见性延迟**（本次事件的根因）：行已插入但 ACE 引擎尚未对其他 ODBC 连接可见（批量插入时窗口可达 11 秒）→ 这时降级 Delete 是**有害的误伤**。
 **代码当前无法区分这两种情况**，统一降级为 Delete。这就是要重构的核心。
 ---
 ## 三、Apply 阶段（数据应用）
 调用存储过程 `usp_SyncApply`，**按表分组、集合化处理**所有 pending 行。
 ```mermaid
 flowchart TD
    A([call_apply<br/>EXEC usp_SyncApply]) --> B["重置 error 行<br/>RetryCount < max_retries 的 → pending"]
    B --> C["按 TargetSchema+TargetTable 分组<br/>遍历每个目标表"]
    C --> D{该表有 pending 行?}
    D -- 否 --> C
    D -- 是 --> E["统计 pending 数 / distinct RecordID 数"]
    E --> F["BEGIN TRAN"]
    F --> G["🔑 构建列清单<br/>从 sys.columns 读目标表所有列<br/>（排除 ID/computed/identity/timestamp）"]
    G --> H["构建动态 SQL"]
    H --> I["分支1: Upsert（MERGE）<br/>ranked CTE: 按 RecordID 分区,<br/>SourceLogID DESC 取 rn=1<br/>仅 OperateType∈Insert/Update 且 RowData 非空"]
    I --> J["SET IDENTITY_INSERT ON<br/>MERGE 目标表<br/>匹配则 UPDATE, 不匹配则 INSERT<br/>@merged = @@ROWCOUNT"]
    J --> K["分支2: Delete<br/>同一 ranked CTE 的 rn=1 行<br/>仅 OperateType=Delete"]
    K --> L["DELETE 目标表 WHERE ID IN (...)<br/>@deleted = @@ROWCOUNT"]
    L --> M["把该表所有 pending 行<br/>Status → applied, AppliedAt = now<br/>@applied = @@ROWCOUNT"]
    M --> N[COMMIT]
    N --> O["写 SyncApplyRunLog 审计<br/>pending/merged/deleted/applied/<br/>error/dead + CycleID + 耗时"]
    O --> C
    N -.失败.-> X["ROLLBACK"]
    X --> Y["超 max_retries → dead<br/>否则 → error（下轮重试）"]
    Y --> O
    style I fill:#e3f2fd,stroke:#1976d2
    style K fill:#ffcdd2,stroke:#c62828
    style M fill:#fff3e0,stroke:#f57c00
 ```
 ### 关键：保序「最后操作胜」（`02_sync_apply.sql:95-135`）
 单个 `ranked` CTE 同时供 Upsert 和 Delete 两个分支使用：
 ```sql
 ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn
 ```
 - `rn=1` 是该 RecordID **真正的最后一条日志**。
 - Upsert 分支：`rn=1 AND OperateType IN ('Insert','Update')`
 - Delete 分支：`rn=1 AND OperateType='Delete'`
 所以**降级成 Delete 的行，在这里会真的去 SQL 端执行 DELETE**。本次事件中 5 行 Delete 的 `@deleted=0`（SQL 里本就没这些行，空打），但 `@applied=5`（队列行照样被标 applied）。
 ---
 ## 四、Cleanup 阶段（日志清除）—— ❗决定「重试」能否成立的命脉
 ```mermaid
 flowchart TD
    A([cleanup 某个文件]) --> B["查 SyncQueue 中<br/>Status=applied 的 SourceLogID 列表<br/>applied_log_ids(file)"]
    B --> C{有 applied 行?}
    C -- 否 --> Z([cleanup 结束, 返回 0])
    C -- 是 --> D["DELETE FROM Access.TableChangeLog<br/>WHERE ID IN (上述列表)<br/>分批 + 锁重试"]
    D --> E{删除数 == 预期?}
    E -- 否 --> F["WARNING: 部分日志已不在<br/>（被外部或中断的运行删过）"]
    E -- 是 --> G["mark_cleaned:<br/>这些队列行 Status → cleaned<br/>CleanedAt = now"]
    F --> G
    G --> H["INFO: 删除了 N 条 Access 日志"]
    H --> Z
    style D fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
    style B fill:#fff9c4,stroke:#f9a825
 ```
 ### 🔑 cleanup 的判定条件是重构的支点
 **cleanup 只删 `Status='applied'` 的日志**（`sql_writer.py:264-272` 的 `applied_log_ids`）。这意味着：
 | capture 对该日志的处理 | SyncQueue 行状态 | cleanup 是否删 Access 日志 | 后果 |
 |------------------------|------------------|----------------------------|------|
 | 降级 Delete（现状） | applied | **删除** | ❌ 日志消失，再无重试机会 |
 | **跳过不入队（重构后）** | （无对应行） | **不删** | ✅ 日志保留，下轮重读 |
 **结论：重构只要做到「读不到 → 不入队」，cleanup 这一段天然会把日志保留下来，无需改动 cleanup.py。** 这是「跨 cycle 重试」能够成立的根基。
 ---
 ## 五、数据流转全景：一条日志的完整生命周期
 ```mermaid
 flowchart LR
    subgraph Access["Access 端（.accdb）"]
        T1[(业务表<br/>如 接收)]
        TCL[(TableChangeLog<br/>变更日志 append-only)]
    end
    subgraph SQL["SQL Server 端（CompanyDB）"]
        SQ[(SyncQueue<br/>待处理队列)]
        ARCH[(SyncLogArchive<br/>永久审计)]
        RL[(SyncApplyRunLog<br/>apply 运行日志)]
        MIRROR[(镜像表<br/>如 接收_YEAR2026)]
    end
    T1 -- "数据宏 After I/U/D<br/>写入一行日志" --> TCL
    TCL -- "① capture 读取" --> CAP[Capture]
    CAP -- "回读整行" --> T1
    CAP -- "② 入队（去重）" --> SQ
    CAP -- "② 审计存档" --> ARCH
    SQ -- "③ apply 处理" --> APPLY[usp_SyncApply]
    APPLY -- "MERGE/DELETE" --> MIRROR
    APPLY -- "pending→applied" --> SQ
    APPLY -- "记录运行结果" --> RL
    SQ -- "④ cleanup 查 applied" --> CLEAN[Cleanup]
    CLEAN -- "⑤ 删已应用日志" --> TCL
    CLEAN -- "applied→cleaned" --> SQ
    style CAP fill:#e3f2fd,stroke:#1976d2
    style APPLY fill:#fff3e0,stroke:#f57c00
    style CLEAN fill:#e8f5e9,stroke:#388e3c
 ```
 ---
 ## 六、本次事件的重放（在上述流程中的路径）
 5 条 Insert 日志（RecordID 16255-16259）在一轮 cycle 中的遭遇：
 ```mermaid
 flowchart TD
    A["8/3 09:01 Access 批量插入 5 行<br/>数据宏写 5 条 Insert 日志<br/>SourceLogID 15533-15537"] --> B
    B["09:00:50 Capture 读取这 5 条日志<br/>（注: CapturedAt 比 OriginalTime 早 11s<br/> = ACE 可见性窗口）"] --> C
    C["read_row 回读 5 行<br/>SELECT * FROM 接收 WHERE ID=16255..16259"] --> D
    D["❌ 全部返回 None<br/>（行尚未对其他连接可见）"] --> E
    E["🔴 降级为 Delete<br/>入队 SyncQueue, OperateType=Delete<br/>RowData=null"] --> F
    F["09:00:51 Apply: Delete 分支<br/>DELETE 接收_YEAR2026 WHERE ID IN(16255..16259)<br/>@deleted=0（SQL 本就没这5行）<br/>@applied=5（队列行标 applied）"] --> G
    G["队列行 Status = applied"] --> H
    H["Cleanup: 查到这5条 applied<br/>DELETE Access.TableChangeLog ID=15533-15537"] --> I
    I["🔴 日志被删，5 条 Insert 证据消失<br/>这5行从此再无机会被同步进 SQL"] --> J
    J["8/4 05:08 compare: missing_in_sql=5<br/>ID 16255-16259"]
    style D fill:#ffcdd2,stroke:#c62828
    style E fill:#ffcdd2,stroke:#c62828,stroke-width:3px
    style H fill:#e8f5e9,stroke:#388e3c
    style I fill:#ffcdd2,stroke:#c62828
    style J fill:#fff9c4,stroke:#f9a825
 ```
 **链式灾难**：降级 Delete（capture）→ 空打 Delete 但标 applied（apply）→ 删 Access 日志（cleanup）。三段配合下，这 5 行被「合法地」从同步链路中抹除。只要在 capture 段断开第一环（不降级、不入队），后面 apply/cleanup 就不会碰它们，日志保留，下轮自然重试。
 ---
 ## 七、重构决策点（待定）
 基于上述流程，重构的核心是改造 capture 阶段的「降级」分支。需要决策的问题：
 1. **读不到时的动作**：跳过不入队（日志保留，下轮重读）vs 入队但标记 defer 状态？
 2. **重试上限的判定依据**：用「重试次数」（需要存储计数）vs 用「日志年龄时间窗」（无需存储，靠 `now - OriginalTime > 阈值`）？
 3. **终态处理**：超限后该 Access 日志保留（占队列头持续重读）还是删除（丢证据）？
 4. **Update 日志**：是否套用同一套 defer 逻辑？
 我倾向的方案：**读不到 → 不入队 + 写 defer 审计；用日志年龄（如 120s）作终态判据；超限则入队标 dead（保留 Access 日志不删，待人工）**。零 schema 改动、零状态存储。等你审完这份流程图确认方向后，我再动手。
--- a/docs/plan-capture-defer-by-age.md
+++ b/docs/plan-capture-defer-by-age.md
@@ -0,0 +1,211 @@
 # 重构方案：capture 读不到行时按「日志年龄」延迟重试
 > 目标：根治「Insert 日志回读不到行 → 降级 Delete → 数据丢失」的缺陷（8/4 事件根因）。
 > 核心改动：读不到 → **不入队、不降级**，按 Access 日志的年龄决定「下轮重试」还是「判死待人工」。
 > 改动面：`capture.py`（主）、`config.py`（新增2个配置项）、`capture.py` 的 CaptureStats（新增计数器）。**零 schema 改动、零 SQL 改动、不动 apply/cleanup。**
 ---
 ## 一、当前问题行为 vs 重构后行为
 | 场景 | 当前行为（缺陷） | 重构后行为 |
 |------|------------------|-----------|
 | Insert/Update 读不到行（可见性延迟，本次事件） | 降级 Delete → 入队 → apply 空打 Delete + 标 applied → cleanup 删 Access 日志 → **数据永久丢失** | 年龄 < 阈值：跳过不入队，日志保留，下轮重读 → 读到后正常 Insert ✅ |
 | Insert/Update 读不到行（真删除：先插后删） | 降级 Delete（碰巧正确） | 年龄 < 阈值：跳过；后续 Delete 日志会兜底正确清理；超龄判死待人工 ✅ |
 | Insert/Update 一直读不到（真异常/数据损坏） | 降级 Delete（错误） | 年龄 ≥ 阈值：入队标 dead，保留 Access 日志，**告警待人工** ✅ |
 ---
 ## 二、配置项新增（`src/sync/config.py` 的 RuntimeConfig）
 ```python
 class RuntimeConfig(BaseModel):
    # ... 既有字段 ...
    # Insert/Update 日志回读不到行时，按日志年龄延迟重试：年龄小于此秒数则
    # 跳过不入队（保留 Access 日志，下一轮 cycle 重新捕获），度过 ACE 引擎
    # 的可见性窗口；超过此年龄仍读不到则判定为真删除/异常，入队标 dead 待人工。
    # 设为 0 可关闭延迟重试（退化为旧的"立即判死"语义，但不再降级 Delete）。
    capture_defer_seconds: int = 120
 ```
 - **默认值 120 秒**：覆盖本次事件观察到的 ~11 秒可见性窗口，并留足 10 倍余量；对 `poll_interval=10s` 意味着最多重试约 12 轮。
 - 单一配置项，`config.yaml` 无需改动即可生效（用默认值）。`defer_dead_op` 不暴露为配置（实现细节，固定为 `dead` 状态，见下）。
 ---
 ## 三、capture.py 改动（核心）
 ### 3.1 CaptureStats 新增两个计数器
 ```python
@dataclass
 class CaptureStats:
    # ... 既有字段 ...
    read: int = 0
    enqueued: int = 0
    dedup_skipped: int = 0
    downgraded: int = 0      # 保留字段，重构后恒为 0（兼容旧日志解析）
    deferred: int = 0        # 【新】读不到行但年龄 < 阈值，跳过待下轮重试
    aged_out: int = 0        # 【新】读不到行且年龄 ≥ 阈值，判死待人工
    out_of_scope: int = 0
    unknown_op: int = 0
    # ... merge() / ops_str() 同步更新 ...
 ```
 ### 3.2 降级分支重构（`capture_file` 中 Insert/Update 回读逻辑）
 **替换** 现有的第 93-108 行（整段 `if op in ("Insert", "Update"):` 块）：
 ```python
 if op in ("Insert", "Update"):
    d = reader.read_row(lr.table_name, lr.record_id)
    if d is None:
        # 读不到行：不再降级 Delete。按 Access 日志年龄决定延迟重试还是判死。
        # 必须计算日志年龄（原始设计无此逻辑）。
        age_seconds = _log_age_seconds(lr.time)
        if age_seconds < cfg.runtime.capture_defer_seconds:
            # 暂态读不到（ACE 可见性窗口）：跳过，不入队、不清理。
            # Access 日志因无对应 applied 队列行，cleanup 不会删除，下轮重试。
            st.deferred += 1
            log.warning(
                "capture DEFER %s file=%s table=%s record_id=%s log_id=%s "
                "log_time=%s age=%ds (source row unreadable; will retry next "
                "cycle while log age < %ds)",
                lr.operate_type, fm.file, lr.table_name, lr.record_id,
                lr.id, lr.time, int(age_seconds),
                cfg.runtime.capture_defer_seconds,
            )
            continue   # ← 关键：跳过本行，archive/queue 都不写
        else:
            # 超龄仍读不到：真删除或真异常。入队标 dead，不降级、不执行任何
            # 破坏性操作。保留 Access 日志（无 applied 行 → cleanup 不删），
            # 队列健康检查会告警，等待人工介入。
            st.aged_out += 1
            log.warning(
                "capture AGED-OUT %s file=%s table=%s record_id=%s log_id=%s "
                "log_time=%s age=%ds >= %ds -- enqueuing as dead for manual "
                "review (source row still unreadable after defer window)",
                lr.operate_type, fm.file, lr.table_name, lr.record_id,
                lr.id, lr.time, int(age_seconds),
                cfg.runtime.capture_defer_seconds,
            )
            op = "dead"  # 仅用于入队时的状态标记，见下
    else:
        row_data = json.dumps(d, ensure_ascii=False)
 ```
 ### 3.3 超龄行的入队方式（`aged_out` 分支）
 超龄行需要进 SyncQueue 但**不能被 apply 执行任何 SQL 操作**（没数据可插，也不能 Delete）。两种实现可选（我倾向 A）：
 **方案 A（推荐）：入队后直接标 dead，OperateType 保留原 Insert/Update 真相**
 - 入队时 `OperateType` 仍写 `Insert`/`Update`（保留原始意图，便于审计），`RowData=null`。
 - 入队后立即 `UPDATE ... SET Status='dead', ErrorMsg='source row unreadable after {age}s defer'`。
 - apply 的游标只选 `Status='pending'`，dead 行不会被处理 → 不会误删。
 - queue 健康检查已有 dead 告警（`service.py:166-184`），自动浮现。
 **方案 B：新增 OperateType='Noop'** — 改动面更大（apply 存储过程需识别），不推荐。
 > 需要在 `sql_writer.py` 新增一个方法 `insert_dead_row(row, error_msg)`，逻辑 = 先 `insert_queue_row`（去重插入）再 `UPDATE ... SET Status='dead', RetryCount=<对应>, ErrorMsg=?`。
 ### 3.4 日志年龄计算辅助函数 `_log_age_seconds`
 ```python
 import datetime as _dt
 def _log_age_seconds(log_time: object) -> float:
    """Access 日志行 Time 字段距今的秒数。log_time 是 pyodbc 返回的 datetime。
    异常时返回一个大数（视为已超龄），确保宁可判死也不无限重试。"""
    try:
        if isinstance(log_time, _dt.datetime):
            return (_dt.datetime.now() - log_time).total_seconds()
        # Access via ODBC 通常返回 datetime；兜底处理 naive/其它类型
        return float("inf")
    except Exception:
        return float("inf")
 ```
 > ⚠️ **时区/时钟注意**：`lr.time` 是 Access 端写入的本地时间，`datetime.now()` 也是本机本地时间，两者同在 114 主机同一时区，可直接相减。本次事件中 OriginalTime/CapturedAt 的"倒挂"现象（差11秒）不影响此逻辑——因为按年龄判断，即便 lr.time 偏早，age 只会被算得更大，倾向判死而非误伤，方向安全。
 ---
 ## 四、归档表（SyncLogArchive）的处理
 | 分支 | 是否写 archive | 理由 |
 |------|---------------|------|
 | DEFER（跳过） | **不写** | 日志保留在 Access，下轮 capture 会重新读到并正常归档；此时写 archive 反而会在去重表里留下"读不到"的半成品记录 |
 | AGED-OUT（判死） | **写** | 超龄是终态，需留永久审计（OriginalOperateType=Insert/Update, ProcessedOperateType='AgedOut', RowData=null, OriginalTime=lr.time） |
 > ProcessedOperateType 新增值 `'AgedOut'`（仅 archive 表用，varchar(10) 放得下 7 字符）。这是纯审计标记，不影响任何执行逻辑。
 ---
 ## 五、不改动的地方（明确边界）
 | 模块 | 是否改动 | 原因 |
 |------|---------|------|
 | `cleanup.py` | ❌ 不改 | 只删 `Status='applied'` 的日志；DEFER 行不入队无 applied 行 → 日志保留；AGED-OUT 标 dead 非 applied → 日志也保留。天然自洽。 |
 | `sql/02_sync_apply.sql` | ❌ 不改 | apply 游标只选 `pending`，dead 行天然跳过；DEFER 行根本不入队。 |
 | `sql/01_sync_queue.sql` | ❌ 不改 | 不新增列，不改索引。 |
 | `service.py` | ❌ 不改 | 队列健康检查已有 error/dead 告警（`queue_error_samples`），AGED-OUT 的 dead 行会自动被它捕获并 WARNING。capture summary 日志格式已包含新计数器（由 CaptureStats.ops_str/merge 驱动）。 |
 | `access_reader.py` | ❌ 不改 | `read_row` 行为不变。 |
 | `config.yaml` | ❌ 不改 | 用默认值 120s 即可。 |
 ---
 ## 六、本次事件 5 行的重放（重构后）
 ```
 cycle N (09:00:50): capture 读到 5 条 Insert 日志
  → read_row 返回 None
  → age = now(09:00:50) - log_time(09:01:01) → 注: 因 Access 时间戳特性 age 可能算成负或小
  → 即便按最保守计算，age 远 < 120s
  → DEFER：跳过不入队，写 WARNING，Access 日志保留
 cycle N+1 (09:01:00): capture 再次读到这 5 条日志
  → read_row 此刻可见性窗口已过（11s > 窗口）→ 读到行 ✅
  → 正常入队 OperateType=Insert，带完整 RowData
  → apply MERGE → SQL 正确写入 5 行 ✅
  → cleanup 删 Access 日志（这次是 applied，合理）
 ```
 **结果：8/4 compare 不再出现 missing_in_sql=5。**
 ---
 ## 七、验证计划
 1. **单元测试**（`tests/test_capture.py`）：
   - mock `read_row` 返回 None + `lr.time` 为近时 → 断言 `deferred=1, enqueued=0, aged_out=0`，不调用 `insert_queue_row` / `insert_archive_row`。
   - mock `read_row` 返回 None + `lr.time` 为 200s 前 → 断言 `aged_out=1`，调用 `insert_dead_row`，Status=dead。
   - mock `read_row` 返回 dict + 任意时间 → 断言正常入队（回归测试）。
   - `capture_defer_seconds=0` → 任何读不到都立即判死（边界）。
 2. **现有测试回归**：`pytest` 全绿（确保去重/正常 Insert/真 Delete 路径不受影响）。
 3. **集成验证**（部署后观察 1-2 天）：
   - 关注 capture summary 日志的 `deferred=` 计数，确认批量插入场景下有 defer 发生且下轮 enqueued。
   - 关注 `queue health` WARNING，确认 dead 行（如有）被正确告警。
   - 跑 `compare --granularity ids`，确认无 missing_in_sql。
 ---
 ## 八、改动文件清单
 | 文件 | 改动类型 | 说明 |
 |------|---------|------|
 | `src/sync/config.py` | 新增字段 | RuntimeConfig 加 `capture_defer_seconds: int = 120` |
 | `src/sync/capture.py` | 核心重构 | 降级分支 → defer/aged_out 分支；CaptureStats 加 2 计数器；新增 `_log_age_seconds` |
 | `src/sync/sql_writer.py` | 新增方法 | `insert_dead_row(row, error_msg)`：去重插入后立即标 dead |
 | `tests/test_capture.py` | 新增用例 | 覆盖 DEFER / AGED-OUT / 正常 / 边界 4 种情况 |
 **总计 4 个文件，零 SQL/零 schema 改动。**
 ---
 ## 九、待你确认的决策点
 1. **`capture_defer_seconds` 默认值 120s** 是否合适？（覆盖11s窗口×10倍余量）
 2. **AGED-OUT 行的处理**：入队标 dead（方案A，推荐）vs 其它？
 3. **archive 表 ProcessedOperateType 新增值 `'AgedOut'`** 是否可接受？
 4. **downgraded 计数器**：保留为恒0（向后兼容旧日志解析）还是直接删除？
 确认后我即按此方案执行。
--- a/docs/plan-health-check-api.md
+++ b/docs/plan-health-check-api.md
@@ -0,0 +1,164 @@
 # 方案：被动式健康检查 API（数据健康 + 服务健康）
 > 目标：在增量同步进程内内置一个轻量 HTTP endpoint，对外暴露结构化健康快照，覆盖「服务健康」与「数据健康」两层。
 > 形态：单进程内置（不另起服务、不新增 NSSM 配置），HTTP server 跑在后台线程，主同步循环零侵入。
 ---
 ## 一、设计原则
 1. **单进程内置**：HTTP server 用标准库 `http.server` + 后台线程，跑在同步进程内。不另起进程、不新增 NSSM 服务、不引入 Flask 等第三方依赖。
 2. **主循环零侵入**：健康检查线程只读共享状态、只读 SQL，不触碰 capture/apply/cleanup 任何路径。HTTP server 崩溃不影响同步。
 3. **数据实时查**：服务健康（最后 cycle 时间/队列状态）从进程内存的状态对象读；数据健康（Access↔SQL 漂移、dead 行）实时查 SQL Server 和最近 compare 报告。
 4. **配置可选**：通过 `config.yaml` 的 `health` 段控制开关/端口/路径，默认开启但端口可配，关闭时不起线程。
 ---
 ## 二、健康检查覆盖的两层
 ### 服务健康（Service Health）
 回答："同步进程在跑吗？最近正常工作吗？队列有没有卡死？"
 | 指标 | 来源 | 含义 |
 |------|------|------|
 | `status` | 综合判定 | healthy / degraded / unhealthy |
 | `pid`, `started_at`, `uptime_seconds` | 进程 | 进程存活与运行时长 |
 | `last_cycle_at`, `last_cycle_duration_s` | 内存状态 | 最近一次 cycle 时间与耗时 |
 | `cycle_lag_seconds` | 实时计算 = now - last_cycle_at | **核心指标**：距上次 cycle 间隔，超过阈值→unhealthy |
 | `last_cycle_active` | 内存状态 | 上轮是否处理了变更 |
 | `queue` | 实时查 SyncQueue | pending/error/dead/cleaned 计数（dead>0→degraded） |
 | `capture_since_start` | 内存状态累计 | 服务启动至今的 deferred/aged_out/enqueued 累计（aged_out>0→degraded） |
 ### 数据健康（Data Health）
 回答："Access 和 SQL 数据一致吗？最近核对结果如何？"
 | 指标 | 来源 | 含义 |
 |------|------|------|
 | `last_compare` | 解析最近的 `compare_ids_*.log` 报告头 | 最近一次 compare 的时间、mismatch 表数、是否一致 |
 | `drift_tables` | 实时查 SyncLogArchive | 启动至今出现过的降级/异常记录涉及的表（AgedOut、Original≠Processed） |
 | `dead_rows_sample` | 实时查 SyncQueue dead 行 | 卡死行的样本（表名/RecordID/错误），辅助定位 |
 > 注：数据健康**不**在每次请求时跑全量 compare（太重），而是读"最近一次定时 compare 的结果" + "归档表/队列里的异常痕迹"。全量漂移检测仍由每日 05:08 的 `compare_ids` 计划任务兜底。
 ---
 ## 三、status 综合判定逻辑
 ```
 unhealthy  : cycle_lag_seconds > 3 × poll_interval  (主循环停滞)
           或 进程内存状态长时间未更新（疑似卡死）
 degraded   : queue.dead > 0          (有放弃的变更，数据可能缺)
           或 queue.error > 0        (有失败待重试)
           或 capture.aged_out > 0   (有读不到的 Insert 被判死)
           或 last_compare.mismatch  (最近核对不一致)
 healthy    : 其余情况
 ```
 `cycle_lag` 用 `3 × poll_interval`（默认 poll=10s → 30s）作阈值：正常空闲 cycle 间隔就是 10s 左右，超过 3 倍说明主循环被卡（比如 Access 锁等待）。
 ---
 ## 四、接口契约
 **请求**：`GET /health`（也可配 `GET /` 简化）
 **响应**：HTTP 200 + JSON（无论 healthy/degraded/unhealthy 都返回 200，状态在 body 里；这样探活失败和网络故障可区分——网络故障是连不上/超时，服务不健康是 200 但 status 字段非 healthy）。
 > 可选：对 unhealthy 同时返回 HTTP 503，便于直接接入按状态码告警的监控平台。这点待定（见决策点）。
 **响应示例**：
 ```json
 {
  "service": "DataMacroSync",
  "status": "healthy",
  "checked_at": "2026-08-05T11:30:00",
  "pid": 1234,
  "started_at": "2026-08-05T11:01:51",
  "uptime_seconds": 1689,
  "service_health": {
    "last_cycle_at": "2026-08-05T11:29:50",
    "last_cycle_duration_s": 1.05,
    "cycle_lag_seconds": 10.0,
    "last_cycle_active": false,
    "queue": {"pending": 0, "error": 0, "dead": 0, "cleaned": 3829},
    "capture_since_start": {"enqueued": 103, "deferred": 0, "aged_out": 0}
  },
  "data_health": {
    "last_compare": {
      "at": "2026-08-05T05:08:21",
      "granularity": "ids",
      "tables_compared": 81,
      "mismatch": false,
      "mismatch_tables": 0
    },
    "drift_tables": [],
    "dead_rows_sample": []
  }
 }
 ```
 ---
 ## 五、实现拆解
 ### 5.1 新增 `src/sync/health.py`（核心，约 200 行）
 - `ServiceState` 数据类：进程内全局状态对象，由 `cycle()` 每轮更新（last_cycle_at / duration / active / 累计计数）。
 - `HealthChecker` 类：持有 `ServiceState` + `SyncConfig`，方法 `snapshot()` 返回上面 JSON 对应的 dict。实时查 SQL 用一次性 SqlWriter（查完即关）。
 - `run_health_server(state, cfg, host, port)`：起 `http.server.ThreadingHTTPServer`，handler 调 `HealthChecker.snapshot()` 序列化返回。
 ### 5.2 改 `src/sync/service.py`
 - `run()` 入口创建 `ServiceState` 实例，传给 `cycle()`；`cycle()` 结束时更新 state（last_cycle_at 等）。
 - `run()` 起健康检查后台线程（`threading.Thread(target=run_health_server, daemon=True)`），主循环照常。
 - 新增累计：把每轮 `CaptureStats` 的 enqueued/deferred/aged_out 累加进 `ServiceState`。
 ### 5.3 改 `src/sync/config.py`
 RuntimeConfig 或新增 HealthConfig：
 ```python
 class HealthConfig(BaseModel):
    enabled: bool = True
    host: str = "0.0.0.0"      # 监听地址，内网可访问
    port: int = 8421            # 健康检查端口
    path: str = "/health"       # URL 路径
 ```
 SyncConfig 增 `health: HealthConfig = HealthConfig()`（默认值，老 config.yaml 无需改动）。
 ### 5.4 改 `config.example.yaml`
 补 `health` 段示例。
 ---
 ## 六、不改动的地方
 | 模块 | 是否改动 | 原因 |
 |------|---------|------|
 | capture/apply/cleanup | ❌ | 主循环逻辑零侵入，只由 cycle 更新一个内存 state |
 | sql_writer 的写方法 | ❌ | 健康检查只用现有的只读查询方法（queue_status_summary 等），不新增写操作 |
 | NSSM 配置 | ❌ | 单进程内置，端口由进程自己起 |
 | main.py | 小改 | `incremental --loop` 分支照常走 `service.run`（健康线程在 run 内起） |
 ---
 ## 七、改动文件清单
 | 文件 | 改动 |
 |------|------|
 | `src/sync/health.py` | **新增** ServiceState + HealthChecker + run_health_server |
 | `src/sync/config.py` | 新增 HealthConfig，SyncConfig 加 health 字段 |
 | `src/sync/service.py` | run() 起 health 线程 + 传 state；cycle() 更新 state |
 | `config.example.yaml` | 补 health 段示例 |
 | `tests/test_health.py` | **新增** 覆盖 status 判定、snapshot 结构、各状态组合 |
 **零 SQL 改动、零 schema 改动、零新第三方依赖（标准库 http.server）。**
 ---
 ## 八、待你确认的决策点
 1. **端口** `8421` 是否合适？（避开 114 上已有服务端口）
 2. **unhealthy 的 HTTP 状态码**：始终 200（状态在 body）vs unhealthy/degraded 返回 503（便于按码告警）？我倾向后者。
 3. **host 监听地址**：`0.0.0.0`（内网/FRP 都可访问）vs `127.0.0.1`（仅本机，需配合 FRP 转发）？我倾向 `0.0.0.0`。
 4. **drift_tables 查询范围**：查"启动至今"的异常归档 vs 查"最近 N 小时"？我倾向"启动至今"（量可控，且能发现历史遗留）。
 确认后即实施。
--- a/docs/plan-health-check-fastapi.md
+++ b/docs/plan-health-check-fastapi.md
@@ -0,0 +1,215 @@
 # 方案 v2：FastAPI 健康检查（面向多接口扩展）
 > 基于"后期会引入更多运维接口"的前提，采用 FastAPI。
 > **与 v1（标准库 http.server）的核心差异在部署模型**——FastAPI/uvicorn 阻塞主线程，必须重新设计进程结构。本方案先解决这个架构问题，再展开实现。
 ---
 ## 一、核心架构决策：同步循环放哪个线程？
 FastAPI 的标准运行方式 `uvicorn.run(app)` 会**阻塞主线程**。而现有项目里 `service.run()`（同步循环）是主线程。两者都要"常驻"，必须有一个让出主线程。这是用 FastAPI 唯一的硬约束，两条路径：
 ### 方案 A：FastAPI 主线程 + 同步循环后台线程（✅ 推荐）
 ```
 NSSM 启动 → main.py incremental --loop
   ├─ 主线程: uvicorn.run(app)         ← FastAPI 常驻
   └─ 后台线程(daemon): service cycle  ← 同步循环搬到后台
 ```
 - **优点**：FastAPI 在主线程，signal handling、uvicorn 内部机制都按官方推荐姿势跑，最稳；未来加接口、加中间件、接 Prometheus 都顺畅。
 - **代价**：同步循环从主线程移到后台线程。但 `cycle()` 本身是纯函数式的（每轮独立、用完即关 writer），搬到后台线程风险可控——它本来就是为"被反复调用"设计的。
 ### 方案 B：同步循环主线程 + FastAPI 后台线程（❌ 不推荐）
 - **缺点**：uvicorn 官方明确不推荐嵌入非主线程，signal handler、事件循环绑定有边角问题；且 NSSM 的进程身份含糊（管的是"同步服务"还是"Web服务"？）。
 **采用方案 A。** 下面所有实现都基于 A。
 ---
 ## 二、进程结构与生命周期
 ```mermaid
 flowchart TD
    NSSM[NSSM 启动 main.py incremental --loop] --> MAIN[主线程]
    MAIN -->|"启动顺序"| S1[1. setup_logging]
    S1 --> S2[2. 创建 ServiceState 全局状态]
    S2 --> S3[3. 启动同步后台线程 daemon]
    S3 --> S4[4. uvicorn.run 主线程阻塞]
    S3 -.->|daemon 线程| CYC[cycle 循环<br/>每轮更新 ServiceState]
    S4 -.->|HTTP 请求到来| APP[FastAPI app]
    APP -->|读| S2
    APP -->|实时查 SQL| DB[(SQL Server)]
    CYC -->|写| S2
    CYC -->|读写| AC[(Access)] & DB
    style S4 fill:#e3f2fd,stroke:#1976d2
    style CYC fill:#fff3e0,stroke:#f57c00
    style APP fill:#e8f5e9,stroke:#388e3c
 ```
 关键点：
 - **ServiceState 是两线程间的唯一桥梁**：同步线程写、HTTP 线程读。用 `threading.Lock` 保护（或用简单的不可变快照替换，避免锁）。
 - **daemon 线程**：同步循环线程设 `daemon=True`，主进程退出时自动终止，不留孤儿。
 - **uvicorn 退出即进程退出**：NSSM stop → uvicorn 收到信号 → 主线程结束 → daemon 同步线程随之终止。
 ---
 ## 三、健康检查覆盖的两层（同 v1）
 ### 服务健康（Service Health）
 | 指标 | 来源 |
 |------|------|
 | `status` | 综合判定 healthy/degraded/unhealthy |
 | `pid`, `started_at`, `uptime_seconds` | 进程 |
 | `last_cycle_at`, `last_cycle_duration_s` | 内存状态（同步线程每轮更新） |
 | `cycle_lag_seconds` | now - last_cycle_at（**核心指标**） |
 | `queue` | 实时查 SyncQueue（pending/error/dead/cleaned） |
 | `capture_since_start` | 内存累计（enqueued/deferred/aged_out） |
 ### 数据健康（Data Health）
 | 指标 | 来源 |
 |------|------|
 | `last_compare` | 解析最近 `compare_ids_*.log` 报告头 |
 | `drift_tables` | 实时查 SyncLogArchive（AgedOut、Original≠Processed） |
 | `dead_rows_sample` | 实时查 SyncQueue dead 行样本 |
 **数据健康不跑全量 compare**（太重），靠"最近定时 compare 结果 + 异常痕迹实时查"。全量漂移仍由每日 05:08 compare 兜底。
 ---
 ## 四、status 综合判定（同 v1）
 ```
 unhealthy : cycle_lag_seconds > 3 × poll_interval   (主循环停滞)
 degraded  : queue.dead > 0 或 queue.error > 0
           或 capture.aged_out > 0
           或 last_compare.mismatch
 healthy   : 其余
 ```
 unhealthy/degraded 时 HTTP 返回 **503**，healthy 返回 200，便于按状态码告警。
 ---
 ## 五、FastAPI 应用结构（为后期扩展铺路）
 ```
 src/sync/
  web/
    __init__.py
    app.py            # FastAPI 实例 + 全局依赖（state/cfg 注入）
    deps.py           # Depends(): 取 ServiceState / SyncConfig
    schemas.py        # Pydantic 响应模型（HealthResponse 等）
    routes/
      __init__.py
      health.py       # GET /health（本次实现）
      # 后期: compare.py (触发/查询核对)、queue.py (死信管理)、metrics.py...
 ```
 - 后期加接口只需在 `routes/` 加文件 + 在 `app.py` `include_router`，结构清晰。
 - 用 FastAPI 的 `Depends` 注入共享的 `ServiceState`，避免全局变量。
 - 响应用 Pydantic 模型（`schemas.py`），自动生成 `/docs` 给运维查阅。
 **本次只实现 `routes/health.py`**，但目录结构一步到位，后期加接口零结构改动。
 ---
 ## 六、配置（config.py 新增）
 ```python
 class HealthConfig(BaseModel):
    enabled: bool = True
    host: str = "0.0.0.0"     # 监听地址
    port: int = 8421           # 健康检查端口
    # uvicorn 运行参数
    log_level: str = "warning" # uvicorn 自身日志级别（避免刷屏）
 class SyncConfig(BaseModel):
    # ... 既有字段 ...
    health: HealthConfig = HealthConfig()  # 默认开启，老 config.yaml 无需改
 ```
 ---
 ## 七、service.py 改造（线程模型变更）
 `run()` 从"主线程跑循环"变为"启动后台同步线程 + 主线程跑 uvicorn"：
 ```python
 def run(cfg):
    setup_logging(cfg.logging)
    state = ServiceState(started_at=datetime.now(), pid=os.getpid())
    # 后台同步线程
    sync_thread = threading.Thread(
        target=_sync_loop, args=(cfg, state), daemon=True, name="sync-cycle"
    )
    sync_thread.start()
    # 主线程跑 FastAPI（阻塞）
    if cfg.health.enabled:
        from sync.web.app import create_app
        import uvicorn
        app = create_app(state, cfg)
        uvicorn.run(app, host=cfg.health.host, port=cfg.health.port,
                    log_level=cfg.health.log_level)
    else:
        # 健康检查关闭：主线程直接跑同步循环（兼容旧行为）
        _sync_loop(cfg, state)
 def _sync_loop(cfg, state):
    """同步循环（原 run() 的 while True 主体，抽出供后台线程调用）。"""
    idle_since = None
    while True:
        t0 = time.monotonic()
        active = cycle(cfg, state)   # cycle 增加 state 参数用于更新状态
        state.update_cycle(active, duration=time.monotonic()-t0)  # 新增
        # idle heartbeat 逻辑保留...
        time.sleep(cfg.runtime.poll_interval_seconds)
 ```
 `cycle(cfg, state)` 末尾新增 `state.update_cycle(...)`，并把每轮 `CaptureStats` 累计进 state。
 ---
 ## 八、依赖更新
 `requirements.txt` 增加：
 ```
 fastapi>=0.110.0
 uvicorn[standard]>=0.27.0
 ```
 （`uvicorn[standard]` 含 uvloop/httptools，性能更好；纯 Windows 下 uvloop 不可用会自动降级，无影响。）
 测试依赖（dev，不入 requirements）：`httpx`（FastAPI 测试用 `TestClient` 依赖它）。
 ---
 ## 九、改动文件清单
 | 文件 | 改动 |
 |------|------|
 | `src/sync/web/app.py` | **新增** create_app + FastAPI 实例 |
 | `src/sync/web/deps.py` | **新增** Depends 注入 |
 | `src/sync/web/schemas.py` | **新增** Pydantic 响应模型 |
 | `src/sync/web/routes/health.py` | **新增** GET /health |
 | `src/sync/health.py` | **新增** ServiceState + HealthChecker（业务逻辑，与 Web 解耦） |
 | `src/sync/service.py` | 改 run() 线程模型 + cycle() 更新 state |
 | `src/sync/config.py` | 新增 HealthConfig |
 | `requirements.txt` | 加 fastapi / uvicorn |
 | `config.example.yaml` | 补 health 段示例 |
 | `tests/test_health.py` | **新增** 覆盖 status 判定/snapshot/TestClient |
 **零 SQL 改动、零 schema 改动。Web 层与业务逻辑（HealthChecker）解耦，后期加接口只动 web/routes/。**
 ---
 ## 十、待确认决策点
 1. **端口 8421**？
 2. **unhealthy/degraded 返回 503**（按码告警友好）确认？
 3. **host 0.0.0.0**（内网+FRP 可访问）确认？
 4. **同步循环放后台 daemon 线程**（方案A）确认？这是与 v1 最大的结构差异。
 5. **drift_tables 查询范围**：启动至今 vs 最近 N 小时？倾向启动至今。
 确认后实施。
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,6 @@
 pyodbc>=5.0.1
 PyYAML>=6.0.1
 pydantic>=2.6.0
 fastapi>=0.110.0
 uvicorn[standard]>=0.27.0
 pytest>=8.0.0
--- a/src/sync/capture.py
+++ b/src/sync/capture.py
@@ -2,23 +2,32 @@
 Every consumed ``TableChangeLog`` row is appended to the permanent audit store
 (``SyncLogArchive``) BEFORE it is enqueued, so the original evidence survives
-cleanup. On top of that, this module now emits a detailed audit trail to the
+cleanup. On top of that, this module emits a detailed audit trail to the
 service log:
- WARNING for every operate-type downgrade (Insert/Update -> Delete because
+- WARNING for every Insert/Update that could not be read back from the source
-  the source row was unreadable at capture time), with record id, log id and
+  table at capture time. Two outcomes, both logged by identity (record id,
-  the Access log timestamp -- the exact event class behind the 14287 incident,
+  log id, log time, age):
-  previously invisible in the text log;
+    * DEFER  -- the log row is younger than ``capture_defer_seconds``: treated
      as an ACE visibility-latency window. The row is NOT enqueued, the Access
      log is left intact (cleanup only deletes rows whose queue status is
      ``applied``), and the next cycle re-reads it. This is the fix for the
      data-loss incident where an Insert downgraded to Delete silently dropped
      rows from SQL Server (see docs/plan-capture-defer-by-age.md).
    * AGED-OUT -- still unreadable past the defer window: enqueued directly as
      ``dead`` (usp_SyncApply never selects dead rows, so no destructive SQL
      runs) and left for manual review; the Access log is also preserved.
 - WARNING for unknown operate types, with enough identity (log id / record id
  / time) to locate and repair the offending log row manually;
 - a per-file INFO summary: rows read, newly enqueued, dedup-skipped
-  (re-capture after a failed apply -- a symptom worth noticing), downgraded,
+  (re-capture after a failed apply -- a symptom worth noticing), deferred,
-  out-of-scope, unknown ops, the processed log-ID range and a per-operation
+  aged-out, out-of-scope, unknown ops, the processed log-ID range and a
-  breakdown;
+  per-operation breakdown;
 - DEBUG detail for individual out-of-scope and dedup skips.
 """
 from __future__ import annotations
 import datetime as _dt
 import json
 import logging
 from dataclasses import dataclass, field
@@ -31,6 +40,23 @@ from .targets import is_synced_table
 log = logging.getLogger(__name__)
 def _log_age_seconds(log_time: object) -> float:
    """Seconds elapsed since the Access log row's ``Time`` value.
    ``log_time`` is the raw pyodbc datetime from ``TableChangeLog.Time`` (set
    by the Access data macro). Both it and ``datetime.now()`` run on the same
    host/clock, so direct subtraction is valid. On any anomaly (missing /
    non-datetime value) returns ``inf`` so the caller errs toward ageing out
    rather than retrying forever.
    """
    try:
        if isinstance(log_time, _dt.datetime):
            return (_dt.datetime.now() - log_time).total_seconds()
        return float("inf")
    except Exception:
        return float("inf")
@dataclass
 class CaptureStats:
    """Counters for one capture pass (per file, or aggregated per cycle)."""
@@ -38,7 +64,8 @@ class CaptureStats:
    read: int = 0            # change-log rows read from Access
    enqueued: int = 0        # rows newly inserted into SyncQueue
    dedup_skipped: int = 0   # already queued (re-capture after a failed apply)
-    downgraded: int = 0      # Insert/Update downgraded to Delete
+    deferred: int = 0        # Insert/Update unreadable but young -> retry next cycle
    aged_out: int = 0        # Insert/Update still unreadable past defer window -> dead
    out_of_scope: int = 0    # log rows for tables outside the sync scope
    unknown_op: int = 0      # log rows with an unrecognised OperateType
    min_log_id: int | None = None
@@ -53,7 +80,8 @@ class CaptureStats:
        self.read += other.read
        self.enqueued += other.enqueued
        self.dedup_skipped += other.dedup_skipped
-        self.downgraded += other.downgraded
+        self.deferred += other.deferred
        self.aged_out += other.aged_out
        self.out_of_scope += other.out_of_scope
        self.unknown_op += other.unknown_op
        for k, v in other.ops.items():
@@ -93,19 +121,66 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
        if op in ("Insert", "Update"):
            d = reader.read_row(lr.table_name, lr.record_id)
            if d is None:
-                # 行已删（或此刻不可读），降级为 Delete。这是数据差异排查的
+                # 回读不到整行：不再降级 Delete。按 Access 日志年龄决定动作，
-                # 头号嫌疑事件（参见 14287 事件），必须在文本日志显式留痕，
+                # 根治"Insert 降级 Delete 致数据丢失"缺陷。
-                # 而不只是写入 SyncLogArchive。
+                age = _log_age_seconds(lr.time)
-                op = "Delete"
+                if age < cfg.runtime.capture_defer_seconds:
-                st.downgraded += 1
+                    # 暂态（ACE 可见性延迟）：跳过，不入队、不写 archive。
                    # Access 日志因无 applied 队列行，cleanup 不会删除，
                    # 下一轮 cycle 会重新读到。
                    st.deferred += 1
                    log.warning(
                        "capture DEFER %s file=%s table=%s record_id=%s "
                        "log_id=%s log_time=%s age=%ds (source row unreadable; "
                        "will retry next cycle while age < %ds)",
                        lr.operate_type, fm.file, lr.table_name, lr.record_id,
                        lr.id, lr.time, int(age),
                        cfg.runtime.capture_defer_seconds,
                    )
                    continue
                # 超龄仍读不到：真删除或真异常。入队标 dead，不降级、不执行
                # 任何破坏性 SQL。保留 Access 日志（无 applied 行 → cleanup
                # 不删），队列健康检查会告警，等待人工介入。
                st.aged_out += 1
                log.warning(
-                    "capture DOWNGRADE %s->Delete file=%s table=%s "
+                    "capture AGED-OUT %s file=%s table=%s record_id=%s "
-                    "record_id=%s log_id=%s log_time=%s "
+                    "log_id=%s log_time=%s age=%ds >= %ds -- enqueuing as "
-                    "(source row unreadable at capture time; original intent "
+                    "dead for manual review (source row still unreadable "
-                    "preserved in SyncLogArchive.OriginalOperateType)",
+                    "after defer window)",
-                    lr.operate_type, fm.file, lr.table_name,
+                    lr.operate_type, fm.file, lr.table_name, lr.record_id,
-                    lr.record_id, lr.id, lr.time,
+                    lr.id, lr.time, int(age),
                    cfg.runtime.capture_defer_seconds,
                )
                target_schema = fm.schema
                target_table = fm.target_table(lr.table_name)
                writer.insert_archive_row(ArchiveRow(
                    source_file=fm.file,
                    source_table=lr.table_name,
                    source_log_id=lr.id,
                    record_id=lr.record_id,
                    target_schema=target_schema,
                    target_table=target_table,
                    original_operate_type=lr.operate_type,
                    processed_operate_type="AgedOut",
                    row_data=None,
                    original_time=lr.time,
                ))
                qr = QueueRow(
                    source_file=fm.file,
                    source_table=lr.table_name,
                    record_id=lr.record_id,
                    target_schema=target_schema,
                    target_table=target_table,
                    source_log_id=lr.id,
                    operate_type=op,   # 保留原始 Insert/Update，便于审计
                    row_data=None,
                )
                writer.insert_dead_row(
                    qr,
                    f"source row unreadable after {int(age)}s defer window "
                    f"(capture_defer_seconds={cfg.runtime.capture_defer_seconds})",
                )
                continue
            else:
                row_data = json.dumps(d, ensure_ascii=False)
        elif op != "Delete":
@@ -122,7 +197,7 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
        # Persist the ORIGINAL log entry to the permanent audit store BEFORE the
        # queue insert (and long before cleanup deletes the Access log). This
        # keeps both the source operate type (lr.operate_type) and the processed
-        # one (op) so a downgrade like Insert->Delete stays reconstructible.
+        # one (op) so any divergence stays reconstructible.
        writer.insert_archive_row(ArchiveRow(
            source_file=fm.file,
            source_table=lr.table_name,
@@ -164,11 +239,11 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
    if st.read:
        log.info(
            "capture file=%s read=%d enqueued=%d dedup_skipped=%d "
-            "downgraded=%d out_of_scope=%d unknown_op=%d "
+            "deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d "
            "log_ids=%s..%s ops={%s}",
-            fm.file, st.read, st.enqueued, st.dedup_skipped, st.downgraded,
+            fm.file, st.read, st.enqueued, st.dedup_skipped, st.deferred,
-            st.out_of_scope, st.unknown_op, st.min_log_id, st.max_log_id,
+            st.aged_out, st.out_of_scope, st.unknown_op, st.min_log_id,
-            st.ops_str(),
+            st.max_log_id, st.ops_str(),
        )
    else:
        log.debug("capture file=%s: change log empty", fm.file)
--- a/src/sync/config.py
+++ b/src/sync/config.py
@@ -27,6 +27,12 @@ class RuntimeConfig(BaseModel):
    cleanup_batch_size: int = 200
    cleanup_lock_retries: int = 3
    cleaned_retention_hours: int = 24
    # Insert/Update 日志回读不到整行时，按 Access 日志的"年龄"（log_time 距今
    # 秒数）决定动作：年龄小于此阈值视为 ACE 引擎的可见性延迟，跳过不入队
    # （Access 日志保留，下一轮 cycle 重新捕获）；年龄达到此阈值仍读不到则
    # 判定为真删除/异常，入队标 dead 待人工。根治"降级 Delete 致数据丢失"
    # （见 docs/plan-capture-defer-by-age.md）。设为 0 关闭延迟重试。
    capture_defer_seconds: int = 60
    # Idle cycles now log at DEBUG; the service emits an INFO heartbeat at this
    # interval while idle so a quiet log still proves the service is alive.
    idle_heartbeat_seconds: int = 600
@@ -47,12 +53,26 @@ class FileMapping(BaseModel):
    def target_table(self, access_table: str) -> str:
        return f"{access_table}{self.year_suffix}"
 class HealthConfig(BaseModel):
    """被动式健康检查 HTTP endpoint 配置（FastAPI/uvicorn）。
    开启时 uvicorn 占主线程、同步循环跑后台 daemon 线程；关闭时退化为
    旧行为（同步循环占主线程）。默认开启，老 config.yaml 无需改动。
    """
    enabled: bool = True
    host: str = "0.0.0.0"      # 监听地址，0.0.0.0 内网/FRP 均可访问
    port: int = 8421           # 健康检查端口
    log_level: str = "warning"  # uvicorn 自身日志级别，避免刷屏
 class SyncConfig(BaseModel):
    sql_server: SqlServerConfig
    access: AccessConfig
    runtime: RuntimeConfig
    files: list[FileMapping]
    logging: dict | None = None
    health: HealthConfig = HealthConfig()
 def load_config(path: str) -> SyncConfig:
    with open(path, "r", encoding="utf-8") as f:
--- a/src/sync/health.py
+++ b/src/sync/health.py
@@ -0,0 +1,308 @@
 """Health-check business logic, decoupled from the web layer.
 Two pieces:
 * :class:`ServiceState` -- the in-memory state shared between the sync thread
  (writer) and the FastAPI thread (reader). Updated by ``service.cycle`` once
  per pass. Lock-free reads are safe because the snapshot is built by copying
  the small set of scalar fields under a short lock; the sync loop never blocks
  on this lock.
 * :class:`HealthChecker` -- builds the structured snapshot dict that the
  ``/health`` endpoint returns. It reads ``ServiceState`` for the live
  process/cycle metrics, then queries SQL Server once for queue health and the
  audit/archive trail for data-health signals. It also parses the most recent
  ``compare_ids_*.log`` report header so data health reflects the last daily
  compare without re-running a full compare on every request (which would be
  far too heavy).
 ``status`` is derived:
 * ``unhealthy`` -- the sync loop appears stalled (cycle lag exceeds a multiple
  of the poll interval, i.e. capture/apply is wedged, typically on an Access
  lock wait);
 * ``degraded`` -- the loop is running but something needs attention: queue
  rows stuck in ``error``/``dead``, an Insert that aged out to dead, or the
  last compare found mismatches;
 * ``healthy`` -- otherwise.
 """
 from __future__ import annotations
 import datetime as _dt
 import logging
 import os
 import re
 import threading
 from .config import SyncConfig
 from .sql_writer import SqlWriter
 log = logging.getLogger(__name__)
 # How stale the last cycle may be (in poll intervals) before status flips to
 # unhealthy. A normal idle cycle paces at poll_interval; 3x tolerates one slow
 # pass (e.g. a large capture batch) without a false alarm.
 CYCLE_LAG_MULTIPLIER = 3
 _COMPARE_HEADER_RE = re.compile(
    r"生成时间\s*:\s*(?P<ts>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})"
    r"|不一致\s*MISMATCH\s*:\s*(?P<mismatch>\d+)"
    r"|一致\s*MATCH\s*:\s*(?P<match>\d+)"
 )
 class ServiceState:
    """In-memory snapshot of the running sync service, updated each cycle.
    Written by the sync thread (in :func:`sync.service.cycle`) and read by the
    health endpoint thread. A single short lock guards the mutable fields; the
    health reader copies them out atomically so it never sees a torn update.
    """
    def __init__(self, started_at: _dt.datetime, pid: int):
        self.started_at = started_at
        self.pid = pid
        self._lock = threading.Lock()
        # Mutable fields (updated each cycle):
        self.last_cycle_at: _dt.datetime | None = None
        self.last_cycle_duration_s: float | None = None
        self.last_cycle_active: bool = False
        self.last_cycle_error: str | None = None
        # Cumulative since process start:
        self.capture_enqueued: int = 0
        self.capture_deferred: int = 0
        self.capture_aged_out: int = 0
    def update_cycle(self, active: bool, duration_s: float,
                     capture=None, error: str | None = None) -> None:
        """Record the outcome of one cycle (called by the sync thread).
        ``capture`` is an optional :class:`sync.capture.CaptureStats`; when
        given its counters are folded into the running totals.
        """
        with self._lock:
            self.last_cycle_at = _dt.datetime.now()
            self.last_cycle_duration_s = duration_s
            self.last_cycle_active = active
            self.last_cycle_error = error
            if capture is not None:
                self.capture_enqueued += capture.enqueued
                self.capture_deferred += capture.deferred
                self.capture_aged_out += capture.aged_out
    def snapshot(self) -> dict:
        """Return a thread-safe copy of the mutable fields (called by reader)."""
        with self._lock:
            return {
                "started_at": self.started_at,
                "pid": self.pid,
                "last_cycle_at": self.last_cycle_at,
                "last_cycle_duration_s": self.last_cycle_duration_s,
                "last_cycle_active": self.last_cycle_active,
                "last_cycle_error": self.last_cycle_error,
                "capture_enqueued": self.capture_enqueued,
                "capture_deferred": self.capture_deferred,
                "capture_aged_out": self.capture_aged_out,
            }
 class HealthChecker:
    """Builds the structured health snapshot for the ``/health`` endpoint."""
    def __init__(self, state: ServiceState, cfg: SyncConfig):
        self.state = state
        self.cfg = cfg
    def snapshot(self) -> dict:
        """Return the full health dict (service_health + data_health + status)."""
        snap = self.state.snapshot()
        now = _dt.datetime.now()
        service = self._service_health(snap, now)
        data = self._data_health()
        status = self._derive_status(service, data)
        return {
            "service": "DataMacroSync",
            "status": status,
            "checked_at": now.isoformat(timespec="seconds"),
            "pid": snap["pid"],
            "started_at": snap["started_at"].isoformat(timespec="seconds"),
            "uptime_seconds": int((now - snap["started_at"]).total_seconds()),
            "service_health": service,
            "data_health": data,
        }
    # ------------------------------------------------------------------ service
    def _service_health(self, snap: dict, now: _dt.datetime) -> dict:
        last = snap["last_cycle_at"]
        cycle_lag = (now - last).total_seconds() if last else None
        return {
            "last_cycle_at": last.isoformat(timespec="seconds") if last else None,
            "last_cycle_duration_s": snap["last_cycle_duration_s"],
            "cycle_lag_seconds": cycle_lag,
            "last_cycle_active": snap["last_cycle_active"],
            "last_cycle_error": snap["last_cycle_error"],
            "queue": self._queue_status(),
            "capture_since_start": {
                "enqueued": snap["capture_enqueued"],
                "deferred": snap["capture_deferred"],
                "aged_out": snap["capture_aged_out"],
            },
        }
    def _queue_status(self) -> dict:
        """Live SyncQueue row counts per status (best-effort)."""
        try:
            w = SqlWriter(
                self.cfg.sql_server.conn_str,
                self.cfg.sql_server.sync_queue_table,
                self.cfg.sql_server.archive_table,
                self.cfg.sql_server.apply_proc,
                self.cfg.sql_server.apply_runlog_table,
            )
            try:
                return w.queue_status_summary()
            finally:
                w.close()
        except Exception as e:
            log.warning("health: queue status query failed: %s", e)
            return {"error": str(e)}
    # -------------------------------------------------------------------- data
    def _data_health(self) -> dict:
        return {
            "last_compare": self._last_compare(),
            "drift_tables": self._drift_tables(),
            "dead_rows_sample": self._dead_rows_sample(),
        }
    def _last_compare(self) -> dict:
        """Parse the most recent compare_ids report header.
        The compare task writes ``logs/compare_ids_<YYYY-MM-DD>.log`` once a
        day; reading its header is far cheaper than re-running a full compare
        on every health request.
        """
        log_dir = self._log_dir()
        try:
            candidates = sorted(
                (f for f in os.listdir(log_dir)
                 if f.startswith("compare_ids_") and f.endswith(".log")),
                reverse=True,
            )
        except OSError:
            return {"available": False}
        if not candidates:
            return {"available": False}
        path = os.path.join(log_dir, candidates[0])
        return self._parse_compare_report(path)
    @staticmethod
    def _parse_compare_report(path: str) -> dict:
        ts = match_count = mismatch_count = None
        try:
            with open(path, "r", encoding="utf-8") as f:
                for line in f:
                    if "生成时间" in line:
                        m = re.search(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", line)
                        if m:
                            ts = m.group(0)
                    elif "一致 MATCH" in line:
                        m = re.search(r":\s*(\d+)", line)
                        if m:
                            match_count = int(m.group(1))
                    elif "不一致 MISMATCH" in line:
                        m = re.search(r":\s*(\d+)", line)
                        if m:
                            mismatch_count = int(m.group(1))
        except OSError as e:
            return {"available": False, "error": str(e)}
        return {
            "available": True,
            "report": os.path.basename(path),
            "at": ts,
            "tables_match": match_count,
            "tables_mismatch": mismatch_count or 0,
            "mismatch": (mismatch_count or 0) > 0,
        }
    def _drift_tables(self) -> list:
        """Tables with abnormal audit records (AgedOut / Original!=Processed).
        Reads SyncLogArchive for rows recorded since this process started --
        a bounded window that surfaces both fresh incidents and any historical
        residue without scanning the whole table.
        """
        try:
            w = SqlWriter(
                self.cfg.sql_server.conn_str,
                self.cfg.sql_server.sync_queue_table,
                self.cfg.sql_server.archive_table,
                self.cfg.sql_server.apply_proc,
                self.cfg.sql_server.apply_runlog_table,
            )
            try:
                cur = w._conn.cursor()
                cur.execute(
                    f"SELECT TargetSchema, TargetTable, ProcessedOperateType, "
                    f"COUNT(*) AS cnt FROM {w.archive_table} "
                    f"WHERE CapturedAt >= ? "
                    f"AND (ProcessedOperateType <> OriginalOperateType "
                    f"     OR ProcessedOperateType = 'AgedOut') "
                    f"GROUP BY TargetSchema, TargetTable, ProcessedOperateType "
                    f"ORDER BY cnt DESC",
                    self.state.started_at,
                )
                return [
                    {"schema": r[0], "table": r[1], "kind": r[2], "count": r[3]}
                    for r in cur.fetchall()
                ]
            finally:
                w.close()
        except Exception as e:
            log.warning("health: drift tables query failed: %s", e)
            return []
    def _dead_rows_sample(self) -> list:
        """A few sample dead/error queue rows for triage."""
        try:
            w = SqlWriter(
                self.cfg.sql_server.conn_str,
                self.cfg.sql_server.sync_queue_table,
                self.cfg.sql_server.archive_table,
                self.cfg.sql_server.apply_proc,
                self.cfg.sql_server.apply_runlog_table,
            )
            try:
                return w.queue_error_samples(5)
            finally:
                w.close()
        except Exception as e:
            log.warning("health: dead rows query failed: %s", e)
            return []
    # ------------------------------------------------------------------ status
    def _derive_status(self, service: dict, data: dict) -> str:
        poll = self.cfg.runtime.poll_interval_seconds
        lag = service.get("cycle_lag_seconds")
        # unhealthy: the sync loop is stalled.
        if lag is not None and lag > CYCLE_LAG_MULTIPLIER * poll:
            return "unhealthy"
        queue = service.get("queue") or {}
        # degraded: running but needs attention.
        if queue.get("dead") or queue.get("error"):
            return "degraded"
        if service.get("capture_since_start", {}).get("aged_out"):
            return "degraded"
        cmp = data.get("last_compare") or {}
        if cmp.get("mismatch"):
            return "degraded"
        return "healthy"
    # ------------------------------------------------------------------ utils
    def _log_dir(self) -> str:
        path = (self.cfg.logging or {}).get("path", "sync.log")
        return os.path.dirname(os.path.abspath(path)) or "."
--- a/src/sync/service.py
+++ b/src/sync/service.py
@@ -26,7 +26,10 @@ in a ``finally``. ``run(cfg)`` loops ``cycle`` with a sleep; ``main()``
 loads the config from ``argv[1]`` (default ``config.yaml``).
 """
 from __future__ import annotations
 import datetime as _dt
 import os
 import sys
 import threading
 import time
 import uuid
 import logging
@@ -36,30 +39,83 @@ from .access_reader import AccessReader
 from .sql_writer import SqlWriter
 from .capture import capture_file, CaptureStats
 from .cleanup import cleanup_file
 from .health import ServiceState
 from .logging_setup import setup_logging, set_cycle_id
 log = logging.getLogger("sync.service")
 def run(cfg):
-    """Run ``cycle`` forever, sleeping ``poll_interval_seconds`` between passes.
+    """Run the sync loop, optionally behind a health-check HTTP server.
-    Configures logging once on entry. Intended to be started by ``main()``
+    Threading model: when ``cfg.health.enabled`` (default), uvicorn occupies
-    under the service host (e.g. NSSM). Not unit-tested (infinite loop);
+    the main thread and the capture/apply/cleanup loop runs on a daemon
-    ``cycle()`` is the testable unit. Emits an idle heartbeat every
+    thread. NSSM sends its stop signal to the main (uvicorn) thread; on exit
-    ``runtime.idle_heartbeat_seconds`` so a quiet log still proves liveness
+    the daemon sync thread is terminated automatically. When health is
-    now that idle cycles log at DEBUG.
+    disabled, the sync loop runs on the main thread (legacy behaviour).
    """
    setup_logging(cfg.logging)
-    log.info(
+    state = ServiceState(started_at=_dt.datetime.now(), pid=os.getpid())
-        "service started: files=%d poll_interval=%ds idle_heartbeat=%ds",
+
-        len(cfg.files), cfg.runtime.poll_interval_seconds,
+    if cfg.health.enabled:
-        cfg.runtime.idle_heartbeat_seconds,
+        # Sync loop on a daemon thread; uvicorn on main.
        sync_thread = threading.Thread(
            target=_sync_loop, args=(cfg, state), daemon=True, name="sync-cycle",
        )
        sync_thread.start()
        log.info(
            "service started: files=%d poll_interval=%ds idle_heartbeat=%ds "
            "(sync loop on daemon thread; health API on %s:%d)",
            len(cfg.files), cfg.runtime.poll_interval_seconds,
            cfg.runtime.idle_heartbeat_seconds, cfg.health.host, cfg.health.port,
        )
        _run_health_server(cfg, state)
    else:
        log.info(
            "service started: files=%d poll_interval=%ds idle_heartbeat=%ds "
            "(health API disabled)",
            len(cfg.files), cfg.runtime.poll_interval_seconds,
            cfg.runtime.idle_heartbeat_seconds,
        )
        _sync_loop(cfg, state)
 def _run_health_server(cfg, state):
    """Start uvicorn on the main thread (blocks until shutdown)."""
    import uvicorn
    from .web.app import create_app
    app = create_app(state, cfg)
    uvicorn.run(
        app,
        host=cfg.health.host,
        port=cfg.health.port,
        log_level=cfg.health.log_level,
    )
 def _sync_loop(cfg, state):
    """The capture -> apply -> cleanup loop, paced by poll_interval.
    Extracted from the legacy ``run`` so it can run on either the main thread
    (health disabled) or a daemon thread (health enabled). Updates ``state``
    each pass so the health endpoint can observe liveness.
    """
    idle_since = None
    idle_cycles = 0
    while True:
-        active = cycle(cfg)
+        t0 = time.monotonic()
        try:
            active, capture = cycle(cfg, state)
        except Exception:
            # cycle() already isolates per-file failures; a top-level exception
            # here means something unexpected -- record it on the state so the
            # health endpoint surfaces it rather than masking a silent stall.
            log.exception("cycle raised unexpectedly")
            state.update_cycle(False, time.monotonic() - t0,
                               error="cycle raised unexpectedly")
            active, capture = False, None
        else:
            state.update_cycle(active, time.monotonic() - t0, capture=capture)
        now = time.monotonic()
        if active:
            idle_since, idle_cycles = None, 0
@@ -74,11 +130,15 @@ def run(cfg):
        time.sleep(cfg.runtime.poll_interval_seconds)
-def cycle(cfg) -> bool:
+def cycle(cfg, state: ServiceState | None = None) -> tuple[bool, CaptureStats | None]:
    """One capture -> apply -> cleanup pass over all files.
-    Returns True when the cycle did any work (captured / applied / cleaned /
+    Returns ``(active, capture_stats)``: ``active`` is True when the cycle did
-    purged anything); ``run`` uses this for idle-heartbeat pacing. Per-file
+    any work (captured / applied / cleaned / purged); ``_sync_loop`` uses this
    for idle-heartbeat pacing and ``state.update_cycle`` for health reporting.
    ``state`` is optional (tests call without it); when given it is NOT updated
    here -- the caller (_sync_loop) updates it once with the final timing, to
    keep the cycle itself free of cross-cutting concerns. Per-file
    capture/cleanup failures are logged and do not abort the cycle. Apply
    failure does not block cleanup. The writer is always closed in a
    ``finally``. Safe to call directly from tests (does not sleep or loop).
@@ -109,10 +169,10 @@ def cycle(cfg) -> bool:
            activity = True
            log.info(
                "capture summary: read=%d enqueued=%d dedup_skipped=%d "
-                "downgraded=%d out_of_scope=%d unknown_op=%d ops={%s}",
+                "deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d ops={%s}",
                total.read, total.enqueued, total.dedup_skipped,
-                total.downgraded, total.out_of_scope, total.unknown_op,
+                total.deferred, total.aged_out, total.out_of_scope,
-                total.ops_str(),
+                total.unknown_op, total.ops_str(),
            )
        else:
            log.debug("capture summary: no new change-log rows in any file")
@@ -214,7 +274,7 @@ def cycle(cfg) -> bool:
        (log.info if activity else log.debug)(
            "cycle finished in %.2fs", time.monotonic() - t0)
-        return activity
+        return activity, total
    finally:
        writer.close()
        set_cycle_id(None)
--- a/src/sync/sql_writer.py
+++ b/src/sync/sql_writer.py
@@ -132,6 +132,35 @@ class SqlWriter:
        # autocommit: statement already committed.
        return cur.rowcount > 0
    def insert_dead_row(self, row: QueueRow, error_msg: str) -> bool:
        """Enqueue ``row`` then immediately mark it ``dead`` (never applied).
        Used by capture for an Insert/Update log whose source row stays
        unreadable past the defer window (``capture_defer_seconds``): the row
        is recorded for audit/health-alerting purposes but never executed by
        ``usp_SyncApply`` (which only selects ``Status='pending'``), so no
        destructive SQL ever runs against the mirror table. The OperateType is
        preserved as the original Insert/Update so the intent stays visible,
        while ErrorMsg explains why it was parked.
        Dedup semantics match ``insert_queue_row``: if the (SourceFile,
        SourceTable, SourceLogID) row already exists, the insert is skipped
        and the existing row is re-marked dead. Returns True when a new row
        was inserted, False on a dedup hit (the re-mark still happens).
        """
        inserted = self.insert_queue_row(row)
        cur = self._conn.cursor()
        cur.execute(
            f"UPDATE {self.queue_table} SET Status='dead', ErrorMsg=? "
            "WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?",
            error_msg,
            row.source_file,
            row.source_table,
            row.source_log_id,
        )
        # autocommit: statement already committed.
        return inserted
    def insert_archive_row(self, row: ArchiveRow) -> bool:
        """Append ``row`` to the permanent audit store (dedup on source keys).
--- a/src/sync/web/init.py
+++ b/src/sync/web/init.py
@@ -0,0 +1,11 @@
 """FastAPI web layer for the sync service.
 Hosts the health-check endpoint and -- later -- additional operational routes
 (compare trigger, dead-letter management, Prometheus metrics, ...). The web
 layer only wires HTTP concerns; all business logic lives in
 :mod:`sync.health` (:class:`HealthChecker`) so it stays testable without a
 running server.
 """
 from .app import create_app
 __all__ = ["create_app"]
--- a/src/sync/web/app.py
+++ b/src/sync/web/app.py
@@ -0,0 +1,33 @@
 """FastAPI application factory.
 ``create_app(state, cfg)`` builds the app, binds the shared singletons (the
 process-wide :class:`ServiceState` and :class:`SyncConfig`) into the dependency
 graph, and mounts route modules. New operational endpoints are added by
 creating a ``routes/<name>.py`` with an ``APIRouter`` and ``include_router``-
 ing it here.
 """
 from __future__ import annotations
 from fastapi import FastAPI
 from ..config import SyncConfig
 from ..health import ServiceState
 from . import deps
 from .routes import health as health_routes
 def create_app(state: ServiceState, cfg: SyncConfig) -> FastAPI:
    """Build the FastAPI app with health state/config bound for injection."""
    app = FastAPI(
        title="DataMacroSync",
        description="Access -> SQL Server incremental sync operational API",
        version="1.0.0",
        # Keep all auto-generated docs under /api alongside the operational
        # routes, so the whole HTTP surface shares one prefix.
        docs_url="/api/docs",
        redoc_url="/api/redoc",
        openapi_url="/api/openapi.json",
    )
    deps.bind(state, cfg)
    app.include_router(health_routes.router)
    return app
--- a/src/sync/web/deps.py
+++ b/src/sync/web/deps.py
@@ -0,0 +1,42 @@
 """Shared FastAPI dependencies.
 The :class:`ServiceState` and :class:`SyncConfig` are created once in
 ``service.run`` and injected into routes via ``Depends``, avoiding global
 singletons. Routes get a ready-to-use :class:`HealthChecker` rather than the
 raw state, so each endpoint declares only what it needs.
 """
 from __future__ import annotations
 from fastapi import Depends
 from ..config import SyncConfig
 from ..health import HealthChecker, ServiceState
 # Bound at app creation time (create_app). Module-level holders are acceptable
 # here because there is exactly one app per process and they are write-once.
 _state: ServiceState | None = None
 _cfg: SyncConfig | None = None
 def bind(state: ServiceState, cfg: SyncConfig) -> None:
    """Stash the singletons for the lifetime of this app (called once)."""
    global _state, _cfg
    _state = state
    _cfg = cfg
 def get_state() -> ServiceState:
    assert _state is not None, "ServiceState not bound -- call bind() first"
    return _state
 def get_config() -> SyncConfig:
    assert _cfg is not None, "SyncConfig not bound -- call bind() first"
    return _cfg
 def get_checker(
    state: ServiceState = Depends(get_state),
    cfg: SyncConfig = Depends(get_config),
 ) -> HealthChecker:
    return HealthChecker(state, cfg)
--- a/src/sync/web/routes/init.py
+++ b/src/sync/web/routes/init.py
@@ -0,0 +1,2 @@
 """Route modules. Add new operational endpoints here and include their router
 in ``app.py``."""
--- a/src/sync/web/routes/health.py
+++ b/src/sync/web/routes/health.py
@@ -0,0 +1,79 @@
 """GET /api/health -- passive health-check endpoint."""
 from __future__ import annotations
 from fastapi import APIRouter, Depends, Response
 from ..deps import get_checker
 from ..schemas import HealthResponse
 from ...health import HealthChecker
 # All operational routes mount under /api (see app.py). The health route lives
 # at /api/health so future endpoints (/api/compare, /api/queue, ...) share one
 # prefix without per-router configuration.
 router = APIRouter(prefix="/api")
@router.get(
    "/health",
    response_model=HealthResponse,
    summary="Get the sync service health snapshot",
    description=(
        "Returns a structured snapshot of the Access → SQL Server incremental "
        "sync service. Use this for passive health probing (uptime monitors, "
        "load-balancer checks, on-call dashboards).\n\n"
        "**What it covers**\n\n"
        "Two layers are reported:\n\n"
        "1. **Service health** — whether the sync loop is alive and making "
        "progress: the time since the last capture/apply/cleanup cycle "
        "(`cycle_lag_seconds`), the current SyncQueue row counts by status, "
        "and cumulative capture counters since process start.\n"
        "2. **Data health** — lightweight signals about Access ↔ SQL Server "
        "consistency: the result of the most recent scheduled full compare "
        "(run daily ~05:08, parsed from its report file — NOT re-run on every "
        "request), plus any abnormal audit traces (rows that aged out, "
        "downgraded operations) and a sample of dead/error queue rows.\n\n"
        "**Status derivation**\n\n"
        "The top-level `status` field is derived from the signals above:\n\n"
        "- `healthy` — the loop is pacing normally and no attention signals "
        "are present.\n"
        "- `degraded` — the loop is running but something needs attention: "
        "queue rows stuck in `error`/`dead`, an Insert that aged out to dead, "
        "or the last compare found mismatches.\n"
        "- `unhealthy` — the sync loop appears stalled (cycle lag exceeds "
        "3× the poll interval, e.g. wedged on an Access lock wait).\n\n"
        "**HTTP status codes**\n\n"
        "The HTTP status mirrors `status` so off-the-shelf monitors that alert "
        "on status code work without parsing JSON:\n\n"
        "- `200 OK` — healthy.\n"
        "- `503 Service Unavailable` — degraded or unhealthy. This keeps "
        "\"service unreachable\" (a network error / timeout, no response at "
        "all) distinguishable from \"service up but needs attention\".\n\n"
        "**Cost**\n\n"
        "Each request performs one short SQL query (queue + audit) and reads "
        "an in-memory state snapshot updated by the sync loop; it does not run "
        "a full compare. Safe to poll at second-scale intervals."
    ),
    responses={
        200: {"description": "Service is healthy."},
        503: {
            "description": (
                "Service is degraded or unhealthy. The body still contains the "
                "full snapshot; inspect `status`, `service_health` and "
                "`data_health` for the cause."
            ),
            "model": HealthResponse,
        },
    },
 )
 def health(response: Response,
           checker: HealthChecker = Depends(get_checker)) -> dict:
    """Compute and return the health snapshot.
    The heavy lifting lives in :class:`sync.health.HealthChecker`; this route
    is a thin HTTP adapter that maps the resulting status to an HTTP code.
    """
    snap = checker.snapshot()
    # Map body status to HTTP code for code-based alerting.
    if snap["status"] != "healthy":
        response.status_code = 503
    return snap
--- a/src/sync/web/schemas.py
+++ b/src/sync/web/schemas.py
@@ -0,0 +1,204 @@
 """Pydantic response models for the API.
 Declaring these (and passing them as ``response_model`` on routes) makes the
 auto-generated OpenAPI/Swagger docs at ``/api/docs`` show the full response
 shape per field, rather than a bare ``object``. Keep them structurally aligned
 with the dicts produced by :class:`sync.health.HealthChecker`.
 """
 from __future__ import annotations
 from pydantic import BaseModel, Field
 class QueueStatus(BaseModel):
    """Live SyncQueue row counts grouped by Status."""
    pending: int | None = Field(
        default=None,
        description="Rows staged but not yet applied by usp_SyncApply. "
                    "Briefly non-zero right after capture; should drain to 0 "
                    "each cycle.",
    )
    error: int | None = Field(
        default=None,
        description="Rows whose apply failed and are under the retry budget. "
                    "Non-zero indicates a transient apply failure; they are "
                    "re-queued automatically. Drives the `degraded` status.",
    )
    dead: int | None = Field(
        default=None,
        description="Rows that exhausted retries (or were parked as dead on "
                    "capture). These represent changes that exist in Access "
                    "but never reached SQL Server and need manual review. "
                    "Drives the `degraded` status.",
    )
    cleaned: int | None = Field(
        default=None,
        description="Rows successfully applied and whose Access change-log "
                    "entry has been removed. Retained for a short audit "
                    "window then purged; not an error signal.",
    )
 class CaptureSinceStart(BaseModel):
    """Cumulative capture counters since the service process started."""
    enqueued: int = Field(
        default=0,
        description="Total change-log rows successfully staged into SyncQueue "
                    "since process start.",
    )
    deferred: int = Field(
        default=0,
        description="Total Insert/Update log rows whose source row was "
                    "momentarily unreadable (ACE visibility latency) and were "
                    "deferred for retry on a later cycle. Occasional non-zero "
                    "values are normal under bulk-insert bursts.",
    )
    aged_out: int = Field(
        default=0,
        description="Total Insert/Update log rows still unreadable past the "
                    "defer window and parked as dead for manual review. "
                    "Non-zero drives the `degraded` status — these rows never "
                    "reached SQL Server.",
    )
 class ServiceHealth(BaseModel):
    """Live process and sync-loop metrics."""
    last_cycle_at: str | None = Field(
        default=None,
        description="ISO 8601 timestamp of the most recent capture→apply→"
                    "cleanup cycle completion. Null until the first cycle "
                    "finishes.",
    )
    last_cycle_duration_s: float | None = Field(
        default=None,
        description="Wall-clock duration of the last cycle, in seconds.",
    )
    cycle_lag_seconds: float | None = Field(
        default=None,
        description="Seconds elapsed since `last_cycle_at`. The primary "
                    "liveness indicator: a healthy idle loop paces at "
                    "poll_interval (default 10s); exceeding 3× poll_interval "
                    "flips the status to `unhealthy`.",
    )
    last_cycle_active: bool = Field(
        default=False,
        description="True if the last cycle actually processed changes "
                    "(captured/applied/cleaned anything), False if it was an "
                    "idle pass.",
    )
    last_cycle_error: str | None = Field(
        default=None,
        description="Error message if the last cycle raised unexpectedly at "
                    "the top level (per-file failures are isolated and do not "
                    "set this). Null when the cycle completed normally.",
    )
    queue: QueueStatus = Field(
        description="Live SyncQueue row counts by status.",
    )
    capture_since_start: CaptureSinceStart = Field(
        description="Cumulative capture counters since process start.",
    )
 class LastCompare(BaseModel):
    """Result of the most recent scheduled full compare (parsed, not re-run)."""
    available: bool = Field(
        default=False,
        description="Whether a compare report file was found. False when no "
                    "report has been generated yet (e.g. before the first "
                    "daily run) or the logs directory is unreadable.",
    )
    report: str | None = Field(
        default=None,
        description="File name of the parsed report (e.g. "
                    "compare_ids_2026-08-05.log).",
    )
    at: str | None = Field(
        default=None,
        description="ISO 8601 timestamp the compare report was generated.",
    )
    tables_match: int | None = Field(
        default=None,
        description="Number of compared tables whose Access and SQL row-ID "
                    "sets matched.",
    )
    tables_mismatch: int | None = Field(
        default=None,
        description="Number of compared tables that diverged. Non-zero drives "
                    "the `degraded` status via `mismatch`.",
    )
    mismatch: bool = Field(
        default=False,
        description="True if any table diverged in the last compare. Drives "
                    "the `degraded` status. Note this reflects the last "
                    "scheduled compare (daily), not a real-time check.",
    )
 class DriftTable(BaseModel):
    """A table with abnormal audit traces since process start."""
    schema: str = Field(description="Target SQL Server schema name.")
    table: str = Field(description="Target table name (with year suffix).")
    kind: str = Field(
        description="Trace kind: 'AgedOut' (Insert unreadable past defer "
                    "window) or the processed operate type when it diverged "
                    "from the original.",
    )
    count: int = Field(description="Number of abnormal audit rows for this table.")
 class DataHealth(BaseModel):
    """Lightweight Access ↔ SQL Server consistency signals."""
    last_compare: LastCompare = Field(
        description="Result of the most recent scheduled full compare, parsed "
                    "from its report file. A full compare is NOT executed on "
                    "every health request (too heavy).",
    )
    drift_tables: list[DriftTable] = Field(
        default_factory=list,
        description="Tables with abnormal audit traces (aged-out or "
                    "downgraded operations) recorded since process start. "
                    "Empty under normal operation.",
    )
    dead_rows_sample: list[dict] = Field(
        default_factory=list,
        description="Up to 5 sample dead/error queue rows for triage "
                    "(table, record id, error message, ...). Empty when the "
                    "queue has no stuck rows.",
    )
 class HealthResponse(BaseModel):
    """Top-level health snapshot returned by GET /api/health."""
    service: str = Field(
        description="Service identifier (always 'DataMacroSync').",
    )
    status: str = Field(
        description="Derived overall status: 'healthy', 'degraded', or "
                    "'unhealthy'. Mirrored to the HTTP status code "
                    "(200 / 503).",
    )
    checked_at: str = Field(
        description="ISO 8601 timestamp the snapshot was generated.",
    )
    pid: int = Field(description="OS process id of the sync service.")
    started_at: str = Field(
        description="ISO 8601 timestamp the service process started.",
    )
    uptime_seconds: int = Field(
        description="Seconds since the service process started.",
    )
    service_health: ServiceHealth = Field(
        description="Live process and sync-loop metrics.",
    )
    data_health: DataHealth = Field(
        description="Lightweight Access ↔ SQL Server consistency signals.",
    )
--- a/tests/test_capture.py
+++ b/tests/test_capture.py
@@ -1,23 +1,34 @@
 import datetime as _dt
 from unittest.mock import MagicMock
 from sync.config import FileMapping, SyncConfig, AccessConfig, RuntimeConfig, SqlServerConfig
 from sync.access_reader import LogRow
 from sync.capture import capture_file
-def _cfg():
+
 def _cfg(defer=60):
    return SyncConfig(sql_server=SqlServerConfig(conn_str="x"),
-                      access=AccessConfig(driver="d", roots={"2026":"r"}),
+                      access=AccessConfig(driver="d", roots={"2026": "r"}),
-                      runtime=RuntimeConfig(),
+                      runtime=RuntimeConfig(capture_defer_seconds=defer),
                      files=[])
 def _fm():
    return FileMapping(file="x.accdb", root="2026", schema="s",
                       year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
 def test_capture_insert_reads_row_and_queues():
    cfg = _cfg()
-    fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
+    fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding",
                     year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
    reader = MagicMock()
-    reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", None)]
+    reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", _dt.datetime.now())]
    reader.read_row.return_value = {"ID": 34041, "订单号": "X1"}
    writer = MagicMock()
-    n = capture_file(fm, reader, writer, cfg)
+    st = capture_file(fm, reader, writer, cfg)
-    assert n == 1
+    assert st.enqueued == 1
    assert st.deferred == 0 and st.aged_out == 0
    args = writer.insert_queue_row.call_args[0][0]
    assert args.target_schema == "TIGWelding"
    assert args.target_table == "表壳焊接记录_YEAR2026"
@@ -25,37 +36,106 @@ def test_capture_insert_reads_row_and_queues():
    assert '"订单号": "X1"' in args.row_data
    assert args.source_log_id == 10
-def test_capture_update_missing_row_downgrades_to_delete():
+
-    cfg = _cfg()
+def test_capture_unreadable_young_row_is_deferred():
-    fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
+    # Insert 日志年龄 < capture_defer_seconds → 跳过，不入队、不写 archive
    cfg = _cfg(defer=60)
    reader = MagicMock()
-    reader.read_log.return_value = [LogRow(11, "T", "5", "Update", None)]
+    reader.read_log.return_value = [LogRow(11, "T", "5", "Insert", _dt.datetime.now())]
-    reader.read_row.return_value = None  # 行已删
+    reader.read_row.return_value = None  # 回读不到
    writer = MagicMock()
-    n = capture_file(fm, reader, writer, cfg)
+    st = capture_file(_fm(), reader, writer, cfg)
-    assert n == 1
+    assert st.deferred == 1
-    args = writer.insert_queue_row.call_args[0][0]
+    assert st.enqueued == 0 and st.aged_out == 0
-    assert args.operate_type == "Delete"
+    writer.insert_queue_row.assert_not_called()
-    assert args.row_data is None
+    writer.insert_dead_row.assert_not_called()
    writer.insert_archive_row.assert_not_called()
 def test_capture_unreadable_young_update_also_deferred():
    # Update 同样走 defer 路径（不降级 Delete）
    cfg = _cfg(defer=60)
    reader = MagicMock()
    reader.read_log.return_value = [LogRow(12, "T", "6", "Update", _dt.datetime.now())]
    reader.read_row.return_value = None
    writer = MagicMock()
    st = capture_file(_fm(), reader, writer, cfg)
    assert st.deferred == 1
    writer.insert_queue_row.assert_not_called()
    writer.insert_dead_row.assert_not_called()
 def test_capture_unreadable_aged_row_marked_dead():
    # Insert 日志年龄 >= capture_defer_seconds → 入队标 dead，不执行破坏性 SQL
    cfg = _cfg(defer=60)
    old_time = _dt.datetime.now() - _dt.timedelta(seconds=200)
    reader = MagicMock()
    reader.read_log.return_value = [LogRow(13, "T", "7", "Insert", old_time)]
    reader.read_row.return_value = None  # 仍读不到
    writer = MagicMock()
    st = capture_file(_fm(), reader, writer, cfg)
    assert st.aged_out == 1
    assert st.enqueued == 0 and st.deferred == 0
    # archive 留痕（ProcessedOperateType=AgedOut）
    arch = writer.insert_archive_row.call_args[0][0]
    assert arch.original_operate_type == "Insert"
    assert arch.processed_operate_type == "AgedOut"
    assert arch.row_data is None
    # 入队标 dead，OperateType 保留原始 Insert
    qr, err = writer.insert_dead_row.call_args[0]
    assert qr.operate_type == "Insert"   # 不降级
    assert qr.row_data is None
    assert "200s" in err
 def test_capture_defer_zero_disables_retry():
    # capture_defer_seconds=0 → 任何读不到都立即判死（边界）
    cfg = _cfg(defer=0)
    reader = MagicMock()
    reader.read_log.return_value = [LogRow(14, "T", "8", "Insert", _dt.datetime.now())]
    reader.read_row.return_value = None
    writer = MagicMock()
    st = capture_file(_fm(), reader, writer, cfg)
    assert st.aged_out == 1
    assert st.deferred == 0
 def test_capture_delete_never_reads_row():
    # Delete 日志无需回读，直接入队
    cfg = _cfg()
    reader = MagicMock()
    reader.read_log.return_value = [LogRow(15, "T", "9", "Delete", _dt.datetime.now())]
    writer = MagicMock()
    st = capture_file(_fm(), reader, writer, cfg)
    assert st.enqueued == 1
    reader.read_row.assert_not_called()
    qr = writer.insert_queue_row.call_args[0][0]
    assert qr.operate_type == "Delete"
    assert qr.row_data is None
 def test_capture_skips_excluded_tables():
    cfg = _cfg()
-    fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"])
+    fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026",
                     exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"])
    reader = MagicMock()
-    reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", None),
+    reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", _dt.datetime.now()),
-                                    LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", None)]
+                                    LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", _dt.datetime.now())]
    writer = MagicMock()
-    assert capture_file(fm, reader, writer, cfg) == 0
+    st = capture_file(fm, reader, writer, cfg)
    assert st.enqueued == 0 and st.out_of_scope == 2
    writer.insert_queue_row.assert_not_called()
 def test_capture_include_tables_filter():
    cfg = _cfg()
    fm = FileMapping(file="x.accdb", root="2026", schema="inspectionRecords", year_suffix="_YEAR2026",
                     exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
    reader = MagicMock()
-    reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", None),
+    reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", _dt.datetime.now()),
-                                    LogRow(2, "其它表", "1", "Insert", None)]
+                                    LogRow(2, "其它表", "1", "Insert", _dt.datetime.now())]
    reader.read_row.return_value = {"ID": 1}
    writer = MagicMock()
-    assert capture_file(fm, reader, writer, cfg) == 1
+    st = capture_file(fm, reader, writer, cfg)
    assert st.enqueued == 1
    assert writer.insert_queue_row.call_args[0][0].target_table == "检验合格记录表_YEAR2026"
--- a/tests/test_health.py
+++ b/tests/test_health.py
@@ -0,0 +1,183 @@
 """Tests for the health-check layer: ServiceState, HealthChecker status logic,
 and the /health endpoint via FastAPI TestClient."""
 import datetime as _dt
 from unittest.mock import MagicMock, patch
 from fastapi.testclient import TestClient
 from sync.config import (
    AccessConfig, FileMapping, HealthConfig, RuntimeConfig,
    SqlServerConfig, SyncConfig,
 )
 from sync.health import CYCLE_LAG_MULTIPLIER, HealthChecker, ServiceState
 from sync.web.app import create_app
 def _cfg(**health_kw):
    return SyncConfig(
        sql_server=SqlServerConfig(conn_str="x"),
        access=AccessConfig(driver="d", roots={"2026": "r"}),
        runtime=RuntimeConfig(poll_interval_seconds=10),
        files=[FileMapping(file="f.accdb", root="2026", schema="s", year_suffix="_Y")],
        health=HealthConfig(**health_kw),
    )
 def _state(**kw):
    s = ServiceState(started_at=_dt.datetime.now(), pid=1)
    # Force last_cycle_at to "recent" by default so cycle_lag is small.
    s.last_cycle_at = _dt.datetime.now()
    for k, v in kw.items():
        setattr(s, k, v)
    return s
 # ---------------------------------------------------------------- ServiceState
 def test_state_snapshot_is_independent_copy():
    s = ServiceState(started_at=_dt.datetime.now(), pid=1)
    s.update_cycle(active=True, duration_s=1.5)
    snap = s.snapshot()
    snap["capture_enqueued"] = 999  # mutate the copy
    assert s.snapshot()["capture_enqueued"] == 0  # original untouched
 def test_state_update_cycle_accumulates_capture_totals():
    from sync.capture import CaptureStats
    s = ServiceState(started_at=_dt.datetime.now(), pid=1)
    s.update_cycle(True, 1.0, capture=CaptureStats(enqueued=5, deferred=2, aged_out=1))
    s.update_cycle(True, 1.0, capture=CaptureStats(enqueued=3, deferred=1, aged_out=0))
    snap = s.snapshot()
    assert snap["capture_enqueued"] == 8
    assert snap["capture_deferred"] == 3
    assert snap["capture_aged_out"] == 1
 # --------------------------------------------------------------- status logic
 def _checker_with(state, cfg, queue=None, compare_mismatch=False):
    """Build a checker whose SQL queries are stubbed (no real DB)."""
    c = HealthChecker(state, cfg)
    c._queue_status = lambda: queue or {}
    c._drift_tables = lambda: []
    c._dead_rows_sample = lambda: []
    c._last_compare = lambda: {"mismatch": compare_mismatch}
    return c
 def test_status_healthy_when_recent_cycle_clean_queue():
    cfg = _cfg()
    state = _state()
    c = _checker_with(state, cfg, queue={"pending": 0, "dead": 0, "error": 0})
    assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()),
                            c._data_health()) == "healthy"
 def test_status_unhealthy_when_cycle_lag_exceeds_threshold():
    cfg = _cfg()  # poll_interval=10 -> threshold = 30s
    # last cycle was 60s ago
    state = ServiceState(started_at=_dt.datetime.now() - _dt.timedelta(seconds=120), pid=1)
    state.last_cycle_at = _dt.datetime.now() - _dt.timedelta(seconds=60)
    c = _checker_with(state, cfg)
    assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()),
                            c._data_health()) == "unhealthy"
 def test_status_degraded_when_dead_rows():
    cfg = _cfg()
    state = _state()
    c = _checker_with(state, cfg, queue={"dead": 2})
    assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()),
                            c._data_health()) == "degraded"
 def test_status_degraded_when_error_rows():
    cfg = _cfg()
    state = _state()
    c = _checker_with(state, cfg, queue={"error": 1})
    assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()),
                            c._data_health()) == "degraded"
 def test_status_degraded_when_aged_out_accumulated():
    cfg = _cfg()
    state = _state()
    state.update_cycle(True, 1.0)  # no capture arg
    state.capture_aged_out = 3     # simulate prior aged-out incidents
    c = _checker_with(state, cfg, queue={})
    snap = c.snapshot()
    assert snap["status"] == "degraded"
 def test_status_degraded_when_compare_mismatch():
    cfg = _cfg()
    state = _state()
    c = _checker_with(state, cfg, queue={}, compare_mismatch=True)
    assert c._derive_status(c._service_health(state.snapshot(), _dt.datetime.now()),
                            c._data_health()) == "degraded"
 # -------------------------------------------------------------- snapshot shape
 def test_snapshot_has_expected_top_level_keys():
    cfg = _cfg()
    state = _state()
    c = _checker_with(state, cfg, queue={"pending": 0})
    snap = c.snapshot()
    for k in ("service", "status", "checked_at", "pid", "started_at",
              "uptime_seconds", "service_health", "data_health"):
        assert k in snap
    assert "cycle_lag_seconds" in snap["service_health"]
    assert "last_compare" in snap["data_health"]
 # ------------------------------------------------------------- HTTP endpoint
 def test_health_endpoint_returns_200_when_healthy():
    cfg = _cfg()
    state = _state()
    with patch.object(HealthChecker, "_queue_status", return_value={}), \
         patch.object(HealthChecker, "_drift_tables", return_value=[]), \
         patch.object(HealthChecker, "_dead_rows_sample", return_value=[]), \
         patch.object(HealthChecker, "_last_compare",
                      return_value={"available": False}):
        app = create_app(state, cfg)
        client = TestClient(app)
        r = client.get("/api/health")
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "healthy"
    assert body["service"] == "DataMacroSync"
 def test_health_endpoint_returns_503_when_unhealthy():
    cfg = _cfg()
    # stale cycle -> unhealthy
    state = ServiceState(started_at=_dt.datetime.now() - _dt.timedelta(seconds=120), pid=1)
    state.last_cycle_at = _dt.datetime.now() - _dt.timedelta(seconds=60)
    with patch.object(HealthChecker, "_queue_status", return_value={}), \
         patch.object(HealthChecker, "_drift_tables", return_value=[]), \
         patch.object(HealthChecker, "_dead_rows_sample", return_value=[]), \
         patch.object(HealthChecker, "_last_compare",
                      return_value={"available": False}):
        app = create_app(state, cfg)
        client = TestClient(app)
        r = client.get("/api/health")
    assert r.status_code == 503
    assert r.json()["status"] == "unhealthy"
 def test_health_endpoint_returns_503_when_degraded():
    cfg = _cfg()
    state = _state()
    with patch.object(HealthChecker, "_queue_status",
                      return_value={"dead": 1}), \
         patch.object(HealthChecker, "_drift_tables", return_value=[]), \
         patch.object(HealthChecker, "_dead_rows_sample", return_value=[]), \
         patch.object(HealthChecker, "_last_compare",
                      return_value={"available": False}):
        app = create_app(state, cfg)
        client = TestClient(app)
        r = client.get("/api/health")
    assert r.status_code == 503
    assert r.json()["status"] == "degraded"
		`@@ -0,0 +1,2 @@`
							`"""Route modules. Add new operational endpoints here and include their router`
							in ``app.py``."""