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
14 changed files with 1338 additions and 15 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/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/config.py
+++ b/src/sync/config.py
@@ -53,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()``
-    under the service host (e.g. NSSM). Not unit-tested (infinite loop);
-    ``cycle()`` is the testable unit. Emits an idle heartbeat every
-    ``runtime.idle_heartbeat_seconds`` so a quiet log still proves liveness
-    now that idle cycles log at DEBUG.
+    Threading model: when ``cfg.health.enabled`` (default), uvicorn occupies
+    the main thread and the capture/apply/cleanup loop runs on a daemon
+    thread. NSSM sends its stop signal to the main (uvicorn) thread; on exit
+    the daemon sync thread is terminated automatically. When health is
+    disabled, the sync loop runs on the main thread (legacy behaviour).
    """
    setup_logging(cfg.logging)
-    log.info(
-        "service started: files=%d poll_interval=%ds idle_heartbeat=%ds",
-        len(cfg.files), cfg.runtime.poll_interval_seconds,
-        cfg.runtime.idle_heartbeat_seconds,
+    state = ServiceState(started_at=_dt.datetime.now(), pid=os.getpid())
+
+    if cfg.health.enabled:
+        # 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 /
-    purged anything); ``run`` uses this for idle-heartbeat pacing. Per-file
+    Returns ``(active, capture_stats)``: ``active`` is True when the cycle did
+    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).
@@ -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/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_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"