fix(txn+security): 仓储改 flush + 启动校验/兜底 + job.error 脱敏 + SSE 异常硬化

P0-1 SqlCredentialStore/save_draft 由自提交改 flush，端点/服务统一 commit （新增 CredentialStore.commit() 统一提交点；token 刷新落库显式提交）；补多凭据一请求中途失败整体回滚集成测试。 P0-2 启动校验 _fernet(enc_key) 快速失败 + catch-all Exception → ErrorEnvelope； credential_enc_key 改 SecretStr。 P0-3 run_job 异常分类：AppError 存 code+message，其余存通用文案不泄 str(exc)。 P0-4 评审/正文 SSE 失败先发 error 事件，尾部 commit 包 try/except。 P1-4 max_version 加 FOR UPDATE 行锁消除 TOCTOU。 P1-5 scan_overdue 谓词下推 + 批量 UPDATE RETURNING。 P1-10 移除 OAuth user_code 日志。 P2 provider_deps 改调网关 build_adapter；accept_service Committable Protocol； CORS 白名单收窄；request_id 安全字符集白名单；stdlib 日志接管；读端点 404 校验； httpx timeout；测试用合法 Fernet key；类型化响应模型(JobResponse/DimensionEntry/ ReviewConflictView/selling_points)+路由 ErrorEnvelope responses（供 codegen）。
2026-06-21 19:32:24 +02:00
parent 2282d4fd24
commit 345cc73965
37 changed files with 737 additions and 115 deletions
--- a/packages/core/tests/test_chapter_version_lock.py
+++ b/packages/core/tests/test_chapter_version_lock.py
@@ -0,0 +1,56 @@
+"""P1-4：`SqlChapterRepo.max_version` 用行锁消除 read-then-insert 的 TOCTOU 竞态。
+
+并发验收同一章时，两请求若都读到同一 max(version) 再各自 +1 插入，会撞
+`(project_id, chapter_no, version)` 唯一约束抛 500。修法：`max_version` 的 SELECT 加
+`with_for_update()` 锁住该章现有行——后到者阻塞到前者提交后才读到更新后的 max。
+
+真正的双事务阻塞行为需真实 PG（由 tests/ 下 E2E 覆盖）；此处以**不连库**的方式断言
+编译出的 SQL 确实带 `FOR UPDATE`（行锁已正确接线），保持单测确定性、不联网。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+import pytest
+from sqlalchemy.dialects import postgresql
+from ww_core.domain.chapter_repo import SqlChapterRepo
+
+PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+
+class _CapturingResult:
+    def scalars(self) -> _CapturingResult:
+        return self
+
+    def __iter__(self) -> Any:
+        return iter([])
+
+
+class _CapturingSession:
+    """fake session：捕获 execute 的 statement，返回空结果（不连库）。"""
+
+    def __init__(self) -> None:
+        self.statements: list[Any] = []
+
+    async def execute(self, statement: Any) -> _CapturingResult:
+        self.statements.append(statement)
+        return _CapturingResult()
+
+
+@pytest.mark.asyncio
+async def test_max_version_query_uses_for_update_row_lock() -> None:
+    session = _CapturingSession()
+    repo = SqlChapterRepo(session)  # type: ignore[arg-type]  # fake 同形（仅用 execute）
+
+    result = await repo.max_version(PROJECT, 1)
+
+    # 无现有行 → 0（read-then-insert 的基线）。
+    assert result == 0
+    # 断言编译出的 SQL 带 FOR UPDATE（行锁），即 with_for_update() 已正确接线（P1-4）。
+    assert len(session.statements) == 1
+    compiled = str(
+        session.statements[0].compile(dialect=postgresql.dialect())  # type: ignore[no-untyped-call]
+    ).upper()
+    assert "FOR UPDATE" in compiled
--- a/packages/core/tests/test_foreshadow_scan_sql.py
+++ b/packages/core/tests/test_foreshadow_scan_sql.py
@@ -0,0 +1,67 @@
+"""P1-5：`SqlForeshadowLedgerRepo.scan_overdue` 谓词下推 + 批量 UPDATE ... RETURNING。
+
+旧实现拉全项目伏笔行到内存再 Python 过滤/逐条 UPDATE；新实现把判据下推为单条
+`UPDATE ... WHERE ... RETURNING`（命中行一次置 OVERDUE）。判据须等价于
+`foreshadow_state.is_overdue`：`expected_close_to IS NOT NULL AND expected_close_to <
+current AND status NOT IN (CLOSED, OVERDUE)`。
+
+真实批量更新行为由 tests/ 下 E2E（真 PG）覆盖；此处以不连库方式断言编译出的 SQL
+确实是带正确谓词 + RETURNING 的 UPDATE（谓词下推已正确接线），保持单测确定性。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+import pytest
+from sqlalchemy.dialects import postgresql
+from ww_core.domain.foreshadow_repo import SqlForeshadowLedgerRepo
+
+PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001")
+
+
+class _CapturingResult:
+    def scalars(self) -> _CapturingResult:
+        return self
+
+    def all(self) -> list[Any]:
+        return []
+
+
+class _CapturingSession:
+    """fake session：捕获 execute 的 statement，返回空 RETURNING 结果（不连库）。"""
+
+    def __init__(self) -> None:
+        self.statements: list[Any] = []
+        self.expired = False
+
+    async def execute(self, statement: Any) -> _CapturingResult:
+        self.statements.append(statement)
+        return _CapturingResult()
+
+    def expire_all(self) -> None:
+        self.expired = True
+
+
+@pytest.mark.asyncio
+async def test_scan_overdue_pushes_predicate_into_update_returning() -> None:
+    session = _CapturingSession()
+    repo = SqlForeshadowLedgerRepo(session)  # type: ignore[arg-type]  # fake 同形
+
+    changed = await repo.scan_overdue(PROJECT, current_chapter=15)
+
+    # 空结果 → 无命中行；且发了一条语句、并 expire_all 同步 ORM 身份。
+    assert changed == []
+    assert session.expired is True
+    assert len(session.statements) == 1
+
+    compiled = str(
+        session.statements[0].compile(dialect=postgresql.dialect())  # type: ignore[no-untyped-call]
+    ).upper()
+    # 谓词下推 + 批量更新 + RETURNING（不再全表扫到内存逐行）。
+    assert "UPDATE" in compiled
+    assert "RETURNING" in compiled
+    assert "EXPECTED_CLOSE_TO" in compiled
+    # CLOSED / OVERDUE 被排除（NOT IN）——命中即真正状态翻转的行。
+    assert "NOT IN" in compiled
--- a/packages/core/ww_core/domain/chapter_repo.py
+++ b/packages/core/ww_core/domain/chapter_repo.py
@@ -13,7 +13,7 @@ import uuid
 from dataclasses import dataclass
 from typing import Protocol

-from sqlalchemy import func, select
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from ww_db.models import Chapter

@@ -48,7 +48,11 @@ class ChapterView:


 class ChapterRepo(Protocol):
-    """章节草稿读写 + 验收晋升接口（按 project_id 隔离）。"""
+    """章节草稿读写 + 验收晋升接口（按 project_id 隔离）。
+
+    写方法（save_draft/promote_to_accepted）**只 flush 不 commit**——提交交调用方
+    （端点/验收事务）统一一次，保证原子性与全仓「仓储只 flush」契约一致。
+    """

    async def save_draft(
        self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1
@@ -116,6 +120,7 @@ class SqlChapterRepo:
        self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1
    ) -> ChapterDraftView:
        # 唯一约束 (project_id, chapter_no, version)；草稿版次固定 → 覆盖同一行，幂等。
+        # **只 flush 不 commit**：提交交调用方（端点）统一一次，与全仓「仓储只 flush」契约一致。
        existing = await self._find_draft(project_id, chapter_no)
        if existing is None:
            row = Chapter(
@@ -127,12 +132,12 @@ class SqlChapterRepo:
                version=DRAFT_VERSION,
            )
            self._s.add(row)
-            await self._s.commit()
+            await self._s.flush()
            await self._s.refresh(row)
            return _to_view(row)
        existing.content = text
        existing.status = DRAFT_STATUS
-        await self._s.commit()
+        await self._s.flush()
        await self._s.refresh(existing)
        return _to_view(existing)

@@ -143,13 +148,21 @@ class SqlChapterRepo:
        return _to_view(row)

    async def max_version(self, project_id: uuid.UUID, chapter_no: int) -> int:
-        result = await self._s.execute(
-            select(func.max(Chapter.version)).where(
-                Chapter.project_id == project_id,
-                Chapter.chapter_no == chapter_no,
+        # 消除 read-then-insert 的 TOCTOU 竞态：`SELECT version ... FOR UPDATE` 锁住该章
+        # 现有行（PG 不允许聚合函数与 FOR UPDATE 并用，故选具体行加锁再 Python 取 max）。
+        # 并发验收同一章时，后到者阻塞到前者提交后才读到更新后的 max（避免撞唯一约束抛 500）。
+        rows = (
+            await self._s.execute(
+                select(Chapter.version)
+                .where(
+                    Chapter.project_id == project_id,
+                    Chapter.chapter_no == chapter_no,
+                )
+                .with_for_update()
            )
-        )
-        return result.scalar_one_or_none() or 0
+        ).scalars()
+        versions = list(rows)
+        return max(versions) if versions else 0

    async def promote_to_accepted(
        self, project_id: uuid.UUID, chapter_no: int, *, content: str, volume: int = 1
--- a/packages/core/ww_core/domain/foreshadow_repo.py
+++ b/packages/core/ww_core/domain/foreshadow_repo.py
@@ -20,13 +20,14 @@ import uuid
 from typing import Any, Protocol

 from pydantic import BaseModel, Field
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from ww_db.models import Foreshadow

 from ww_core.domain.foreshadow_state import (
+    CLOSED,
+    OVERDUE,
    ForeshadowStatus,
-    apply_overdue_scan,
 )
 from ww_core.domain.foreshadow_state import (
    transition as apply_transition,
@@ -201,19 +202,24 @@ class SqlForeshadowLedgerRepo:
    async def scan_overdue(
        self, project_id: uuid.UUID, *, current_chapter: int
    ) -> list[ForeshadowLedgerView]:
-        rows = (
-            await self._s.execute(select(Foreshadow).where(Foreshadow.project_id == project_id))
-        ).scalars()
-        changed: list[ForeshadowLedgerView] = []
-        for row in rows:
-            new = apply_overdue_scan(
-                ForeshadowStatus(row.status),
-                current_chapter=current_chapter,
-                expected_close_to=row.expected_close_to,
-            ).value
-            if new != row.status:
-                row.status = new
-                changed.append(_to_view(row))
-        if changed:
-            await self._s.flush()
-        return changed
+        # 谓词下推 + 批量 UPDATE ... RETURNING（不再拉全表到内存逐行过滤/更新，P1-5）。
+        # 判据与 `foreshadow_state.is_overdue` 一致：expected_close_to 非空且
+        # current_chapter > expected_close_to 且当前状态既非 CLOSED（终态）也非 OVERDUE
+        # （已逾期则无变更）——后两者排除使本批量更新等价于「只改真正发生状态翻转的行」。
+        stmt = (
+            update(Foreshadow)
+            .where(
+                Foreshadow.project_id == project_id,
+                Foreshadow.expected_close_to.is_not(None),
+                Foreshadow.expected_close_to < current_chapter,
+                Foreshadow.status.not_in([CLOSED.value, OVERDUE.value]),
+            )
+            .values(status=OVERDUE.value)
+            .returning(Foreshadow)
+        )
+        rows = (await self._s.execute(stmt)).scalars().all()
+        # 先把命中行物化为不可变 View（脱离 ORM 身份），再 expire——避免 UPDATE...RETURNING
+        # 与 session 里已加载实例的状态不一致（后续读重取最新）。
+        views = [_to_view(row) for row in rows]
+        self._s.expire_all()
+        return views