fix: 增强 SQLite 数据库恢复与异常提示

app/db.py

@@ -1,7 +1,7 @@
 from collections.abc import Generator
 from pathlib import Path
 
-from sqlalchemy import create_engine, event
+from sqlalchemy import Engine, create_engine, event
 from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
 
 from app.config import get_settings
@@ -34,10 +34,41 @@ engine = create_sqlite_engine(get_settings().database_url)
 SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
 
 
+def check_database_integrity(target_engine: Engine = engine) -> str:
+    raw_connection = target_engine.raw_connection()
+    try:
+        cursor = raw_connection.driver_connection.cursor()
+        try:
+            cursor.execute("PRAGMA integrity_check")
+            result = cursor.fetchone()
+        finally:
+            cursor.close()
+    finally:
+        raw_connection.close()
+
+    status = result[0] if result else "missing_result"
+    if status != "ok":
+        raise RuntimeError(f"SQLite integrity check failed: {status}")
+    return status
+
+
+def checkpoint_sqlite_wal(target_engine: Engine = engine) -> bool:
+    if str(target_engine.url) == "sqlite:///:memory:":
+        return False
+    if target_engine.url.get_backend_name() != "sqlite":
+        return False
+
+    with target_engine.begin() as connection:
+        connection.exec_driver_sql("PRAGMA wal_checkpoint(TRUNCATE)")
+    return True
+
+
 def init_db() -> None:
     import app.models  # noqa: F401
 
     Base.metadata.create_all(engine)
+    check_database_integrity(engine)
+    checkpoint_sqlite_wal(engine)
 
 
 def get_db_session() -> Generator[Session]:

docs/MVP-WorkOrders.md

@@ -532,6 +532,28 @@ fix: 完善失败重试与跳过边界
 fix: 增强 SQLite 数据库恢复与异常提示
 ```
 
+完成记录：
+
+```text
+完成日期：2026-07-03
+相关 commit：fix: 增强 SQLite 数据库恢复与异常提示
+验证命令：
+- .venv/bin/python -m pytest tests/unit/test_db_stability.py tests/integration/test_task_recovery.py -q
+- docker compose up -d --build
+- curl -f http://localhost:8000/health
+- docker compose exec -T app python 检查 PRAGMA integrity_check 与 wal_checkpoint(TRUNCATE)
+- .venv/bin/python -m pytest tests/unit tests/integration -q
+- .venv/bin/python -m pytest tests/unit tests/integration --cov=app --cov-branch --cov-report=term-missing
+验收结论：
+- 启动时执行 PRAGMA integrity_check，异常时抛出明确 RuntimeError，不新增配置开关。
+- 启动建表后执行 PRAGMA wal_checkpoint(TRUNCATE)，将 WAL 写入主库并截断 WAL 文件。
+- Docker 重建后，独立脚本可读到 WO-04 默认规模任务 4bdf36df-8ae4-4059-af98-602f4872dbaa 与 378b173a-a70f-4da1-b235-03e7632e614a。
+- Uvicorn 进程不再持有 deleted app.db-wal/app.db-shm 文件句柄。
+- Docker 重启后 running 任务恢复逻辑仍通过测试。
+遗留问题：
+- 数据库损坏备份策略和本地重置说明需要在 WO-09 用户操作文档中补充。
+```
+
 ### WO-07 导出链路验收与文件质量优化
 
 优先级：P1

tests/unit/test_db_stability.py

@@ -0,0 +1,65 @@
+import pytest
+
+from app.db import check_database_integrity, checkpoint_sqlite_wal, create_sqlite_engine
+
+
+def test_check_database_integrity_returns_ok_for_valid_sqlite_database():
+    engine = create_sqlite_engine("sqlite:///:memory:")
+    try:
+        assert check_database_integrity(engine) == "ok"
+    finally:
+        engine.dispose()
+
+
+def test_check_database_integrity_raises_when_sqlite_reports_problem(monkeypatch):
+    class FakeCursor:
+        def execute(self, _sql):
+            return None
+
+        def fetchone(self):
+            return ("database disk image is malformed",)
+
+        def close(self):
+            return None
+
+    class FakeConnection:
+        def cursor(self):
+            return FakeCursor()
+
+    class FakeRawConnection:
+        driver_connection = FakeConnection()
+
+        def close(self):
+            return None
+
+    class FakeEngine:
+        def raw_connection(self):
+            return FakeRawConnection()
+
+    with pytest.raises(RuntimeError, match="SQLite integrity check failed"):
+        check_database_integrity(FakeEngine())
+
+
+def test_checkpoint_sqlite_wal_truncates_wal_for_file_database(tmp_path):
+    db_path = tmp_path / "app.db"
+    engine = create_sqlite_engine(f"sqlite:///{db_path}")
+    try:
+        with engine.begin() as connection:
+            connection.exec_driver_sql("CREATE TABLE demo (id INTEGER PRIMARY KEY, name TEXT)")
+            connection.exec_driver_sql("INSERT INTO demo (name) VALUES ('ok')")
+
+        assert checkpoint_sqlite_wal(engine) is True
+
+        with engine.connect() as connection:
+            rows = connection.exec_driver_sql("SELECT name FROM demo").all()
+        assert rows == [("ok",)]
+    finally:
+        engine.dispose()
+
+
+def test_checkpoint_sqlite_wal_skips_in_memory_database():
+    engine = create_sqlite_engine("sqlite:///:memory:")
+    try:
+        assert checkpoint_sqlite_wal(engine) is False
+    finally:
+        engine.dispose()