feat(server): 工程骨架后端——上传 → 解析 → 清单最小链路

- core：配置（prod 拒绝默认密钥 / dev 短信 / 假模型）、租户事务 + 行级安全策略、
  problem+json 错误码、信封加密与签名下载、对象存储、事务内入队、Valkey stream 事件流、权限点
- identity：手机号 HMAC 存储、注册、登录（挤下线需确认，API3）、服务端会话
- billing：额度流水只追加，预扣 / 结算 / 退回，一笔预扣只有一个结局
- document：分片上传、按文件头识别格式、Gotenberg 转换、文字层与稳定句子编号
- checklist：标注规则识别、模型抽取、三道校验、标注类条款规则兜底、重跑对齐
- ai：提示词注册表（extract.substantive v1 来自 S4 v2）、假模型、OpenAI 兼容供应商、ai_call 流水
- 迁移：data-model 基线、套餐、Procrastinate、会话状态函数
- 测试 30 个：租户隔离、额度、失败退回、对象加密、契约一致性
- Compose（含 migrate）、镜像、CI

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server/src/bid_doctor/document/api.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import json
+from typing import Annotated, Any, Literal
+from uuid import UUID
+
+from fastapi import APIRouter, Body, Query, Request
+from fastapi.responses import Response
+from pydantic import BaseModel, Field
+from sqlalchemy import select
+
+from bid_doctor.core import keys
+from bid_doctor.core.config import get_settings
+from bid_doctor.core.crypto import open_object, sign_download, verify_download
+from bid_doctor.core.db import TenantContext, tenant_tx
+from bid_doctor.core.errors import AppError
+from bid_doctor.core.storage import storage
+from bid_doctor.document import service, uploads
+from bid_doctor.document.models import DocumentPage, SourceDocument
+from bid_doctor.identity.deps import Ctx
+
+router = APIRouter(tags=["uploads", "documents"])
+
+
+class UploadIn(BaseModel):
+    filename: str = Field(max_length=255)
+    size_bytes: int = Field(ge=1)
+    sha256: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$")
+    purpose: Literal["project_document", "qualification_file"]
+
+
+class DocumentIn(BaseModel):
+    upload_id: UUID
+    role: Literal["tender", "addendum", "clarification", "bid_volume"]
+    volume_name: str | None = Field(default=None, max_length=40)
+
+
+class ParseIn(BaseModel):
+    expected_charge: int | None = None
+
+
+@router.post("/uploads", status_code=201)
+async def create_upload(body: UploadIn, ctx: Ctx) -> dict[str, Any]:
+    return await uploads.create(ctx, body.filename, body.size_bytes, body.purpose, body.sha256)
+
+
+@router.get("/uploads/{upload_id}")
+async def get_upload(upload_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    return await uploads.describe(ctx, upload_id)
+
+
+@router.put("/uploads/{upload_id}/parts/{part_no}", status_code=204)
+async def put_part(upload_id: UUID, part_no: int, request: Request, ctx: Ctx) -> None:
+    await uploads.put_part(ctx, upload_id, part_no, await request.body())
+
+
+@router.post("/uploads/{upload_id}/complete")
+async def complete_upload(upload_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    return await uploads.complete(ctx, upload_id)
+
+
+@router.delete("/uploads/{upload_id}", status_code=204)
+async def abort_upload(upload_id: UUID, ctx: Ctx) -> None:
+    await uploads.discard(ctx, upload_id)
+
+
+@router.get("/projects/{project_id}/documents")
+async def list_documents(project_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        docs = (
+            (
+                await s.execute(
+                    select(SourceDocument)
+                    .where(SourceDocument.project_id == project_id)
+                    .order_by(SourceDocument.created_at)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        return {"items": [await service.doc_to_api(s, d) for d in docs]}
+
+
+@router.post("/projects/{project_id}/documents", status_code=201)
+async def create_document(project_id: UUID, body: DocumentIn, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        doc = await service.create(s, ctx, project_id, body.upload_id, body.role, body.volume_name)
+        return await service.doc_to_api(s, doc)
+
+
+@router.get("/documents/{document_id}")
+async def get_document(document_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return await service.doc_to_api(s, await service.get(s, document_id))
+
+
+@router.get("/documents/{document_id}/pages")
+async def list_pages(document_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        doc = await service.get(s, document_id)
+        pages = (
+            (
+                await s.execute(
+                    select(DocumentPage).where(DocumentPage.document_id == doc.id).order_by(DocumentPage.page_no)
+                )
+            )
+            .scalars()
+            .all()
+        )
+    return {
+        "normalized_sha256": doc.normalized_sha256.hex() if doc.normalized_sha256 else None,
+        "items": [
+            {
+                "page_no": p.page_no,
+                "width_pt": p.width_pt,
+                "height_pt": p.height_pt,
+                "rotation": p.rotation,
+                "text_source": p.text_source,
+            }
+            for p in pages
+        ],
+    }
+
+
+@router.get("/documents/{document_id}/pages/{page_no}/sentences")
+async def list_sentences(document_id: UUID, page_no: int, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        doc = await service.get(s, document_id)
+        dk = await keys.data_key(s, ctx, doc.project_id)
+    if not doc.text_layer_key:
+        raise AppError("document_not_ready")
+    pages = json.loads(open_object(dk, storage().get(doc.text_layer_key), doc.text_layer_key))
+    page = next((p for p in pages if p["page_no"] == page_no), None)
+    if page is None:
+        raise AppError("not_found")
+    return {"items": page["sentences"]}
+
+
+@router.get("/documents/{document_id}/file")
+async def get_file_url(
+    document_id: UUID, ctx: Ctx, variant: Literal["normalized", "original"] = Query()
+) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        doc = await service.get(s, document_id)
+    if variant == "normalized" and not doc.normalized_key:
+        raise AppError("document_not_ready")
+    payload = f"{ctx.tenant_id}:{doc.id}:{variant}"
+    sig, exp = sign_download(payload, get_settings().download_url_ttl_s)
+    from datetime import UTC, datetime
+
+    url = f"{get_settings().public_base_url}/api/v1/files/{ctx.tenant_id}/{doc.id}/{variant}?exp={exp}&sig={sig}"
+    return {"url": url, "expires_at": datetime.fromtimestamp(exp, UTC)}
+
+
+@router.get("/files/{tenant_id}/{document_id}/{variant}", include_in_schema=False)
+async def download(
+    tenant_id: UUID, document_id: UUID, variant: Literal["normalized", "original"], exp: int, sig: str
+) -> Response:
+    """签名下载（本地磁盘驱动用）。不带鉴权头，靠短时签名；S3 驱动时直接返回对象存储的预签名地址。"""
+    if not verify_download(f"{tenant_id}:{document_id}:{variant}", exp, sig):
+        raise AppError("not_found")
+    ctx = TenantContext(tenant_id=tenant_id, user_id=UUID(int=0))
+    async with tenant_tx(ctx) as s:
+        doc = await service.get(s, document_id)
+        dk = await keys.data_key(s, ctx, doc.project_id)
+    key = doc.normalized_key if variant == "normalized" else doc.original_key
+    assert key is not None
+    media = "application/pdf" if variant == "normalized" else "application/octet-stream"
+    return Response(
+        open_object(dk, storage().get(key), key), media_type=media, headers={"Cache-Control": "private, max-age=300"}
+    )
+
+
+@router.post("/documents/{document_id}/parse-jobs", status_code=201)
+async def start_parse(document_id: UUID, ctx: Ctx, body: Annotated[ParseIn | None, Body()] = None) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        job = await service.start_parse(s, ctx, document_id, body.expected_charge if body else None)
+        return service.job_to_api(job)
+
+
+@router.get("/parse-jobs/{job_id}")
+async def get_parse_job(job_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return service.job_to_api(await service.get_job(s, job_id))
+
+
+@router.post("/parse-jobs/{job_id}/cancel", status_code=202)
+async def cancel_parse_job(job_id: UUID, ctx: Ctx) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        return service.job_to_api(await service.request_cancel(s, job_id))

server/src/bid_doctor/document/citations.py

@@ -0,0 +1,59 @@
+"""出处（data-model §5.1）：不可变，清单项、问题、项目字段、标注规则都引用它。"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from bid_doctor.core.db import TenantContext
+from bid_doctor.core.ids import uuid7
+from bid_doctor.document.models import Citation, SourceDocument
+from bid_doctor.document.pipeline.text_layer import Sentence
+
+LABELS = {"tender": "招标文件", "addendum": "补遗", "clarification": "澄清", "bid_volume": "投标文件"}
+
+
+def document_label(role: str, seq_no: int) -> str:
+    return f"{LABELS[role]} {seq_no}" if role in ("addendum", "clarification") else LABELS.get(role, role)
+
+
+async def from_sentences(
+    s: AsyncSession, ctx: TenantContext, doc: SourceDocument, sentences: list[Sentence], kind: str, checks: list[str]
+) -> Citation:
+    """一条出处只能在同一页上：跨页的要求取第一页的句子（跨页拆成多条出处，TODO）。"""
+    assert doc.normalized_sha256 is not None
+    page = sentences[0].page_no
+    same = [x for x in sentences if x.page_no == page]
+    c = Citation(
+        id=uuid7(),
+        tenant_id=ctx.tenant_id,
+        project_id=doc.project_id,
+        document_id=doc.id,
+        normalized_sha256=doc.normalized_sha256,
+        page_no=page,
+        sentence_keys=[x.key for x in same],
+        quads=[q for x in same for q in x.quads],
+        quote="".join(x.text for x in same),
+        created_by_kind=kind,
+        checks_passed=checks,
+    )
+    s.add(c)
+    await s.flush()
+    return c
+
+
+def to_api(c: Citation, label: str) -> dict[str, Any]:
+    return {
+        "id": c.id,
+        "document_id": c.document_id,
+        "document_label": label,
+        "normalized_sha256": c.normalized_sha256.hex(),
+        "page_no": c.page_no,
+        "sentence_keys": c.sentence_keys,
+        "char_start": c.char_start,
+        "char_end": c.char_end,
+        "quote": c.quote,
+        "quads": c.quads,
+        "created_by_kind": c.created_by_kind,
+    }

server/src/bid_doctor/document/formats.py

@@ -0,0 +1,64 @@
+"""按文件头识别真实格式，不看扩展名（FR-M2-01、07 F2）。文件是不可信输入：只读结构，不执行任何内容。"""
+
+from __future__ import annotations
+
+import io
+import zipfile
+from dataclasses import dataclass
+
+import pypdfium2 as pdfium
+
+from bid_doctor.core.errors import AppError
+
+OLE_MAGIC = bytes.fromhex("D0CF11E0A1B11AE1")
+EXTENSIONS = {"pdf": {"pdf"}, "docx": {"docx", "doc", "wps"}, "doc": {"doc", "wps"}, "ofd": {"ofd"}}
+
+
+@dataclass(frozen=True, slots=True)
+class Detected:
+    format: str  # pdf / docx / doc / ofd
+    page_count: int | None  # 只有 PDF 能直接读出页数
+    extension_mismatch: bool
+
+
+def detect(data: bytes, filename: str) -> Detected:
+    ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+    if data.startswith(b"%PDF"):
+        fmt, pages = "pdf", _pdf_pages(data)
+    elif data.startswith(b"PK\x03\x04"):
+        fmt, pages = _zip_kind(data), None
+    elif data.startswith(OLE_MAGIC):
+        # 旧版 DOC 和 WPS 都是 OLE 复合文档；转换时一样交给 LibreOffice。加密的 DOC 在转换阶段报错
+        fmt, pages = "doc", None
+    else:
+        raise AppError("file_format_unsupported", "支持 PDF、Word（DOCX / DOC）和 WPS 文件")
+    if fmt == "ofd":
+        raise AppError("file_format_unsupported", "暂不支持 OFD 版式文件，请上传 PDF 或 Word")
+    return Detected(format=fmt, page_count=pages, extension_mismatch=ext not in EXTENSIONS.get(fmt, set()))
+
+
+def _pdf_pages(data: bytes) -> int:
+    try:
+        doc = pdfium.PdfDocument(data)
+    except pdfium.PdfiumError as e:
+        if "password" in str(e).lower():
+            raise AppError("file_encrypted") from e
+        raise AppError("file_corrupted") from e
+    try:
+        return len(doc)
+    finally:
+        doc.close()
+
+
+def _zip_kind(data: bytes) -> str:
+    try:
+        names = set(zipfile.ZipFile(io.BytesIO(data)).namelist())
+    except zipfile.BadZipFile as e:
+        raise AppError("file_corrupted") from e
+    if "word/document.xml" in names:
+        return "docx"
+    if "OFD.xml" in names:
+        return "ofd"
+    if "EncryptionInfo" in names or "EncryptedPackage" in names:
+        raise AppError("file_encrypted")
+    raise AppError("file_format_unsupported", "支持 PDF、Word（DOCX / DOC）和 WPS 文件")

server/src/bid_doctor/document/models.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from sqlalchemy import FetchedValue
+from sqlalchemy.orm import Mapped, mapped_column
+
+from bid_doctor.core.columns import PK, Bytes, DbTs, Json, TextArray, Ts, UuidCol
+from bid_doctor.core.db import Base
+
+
+class SourceDocument(Base):
+    __tablename__ = "source_document"
+    id: Mapped[PK]
+    tenant_id: Mapped[UuidCol]
+    project_id: Mapped[UuidCol]
+    role: Mapped[str]
+    seq_no: Mapped[int]
+    volume_name: Mapped[str | None]
+    lineage_id: Mapped[UuidCol]
+    original_filename: Mapped[str]
+    detected_format: Mapped[str]
+    size_bytes: Mapped[int]
+    original_key: Mapped[str]
+    original_sha256: Mapped[Bytes]
+    normalized_key: Mapped[str | None]
+    normalized_sha256: Mapped[Bytes | None]
+    page_count: Mapped[int | None]
+    text_layer_key: Mapped[str | None]
+    pipeline_profile: Mapped[Json | None]
+    status: Mapped[str] = mapped_column(server_default=FetchedValue())
+    uploaded_by: Mapped[UuidCol]
+    created_at: Mapped[DbTs]
+
+
+class DocumentPage(Base):
+    __tablename__ = "document_page"
+    tenant_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    document_id: Mapped[UuidCol] = mapped_column(primary_key=True)
+    page_no: Mapped[int] = mapped_column(primary_key=True)
+    width_pt: Mapped[float]
+    height_pt: Mapped[float]
+    rotation: Mapped[int]
+    text_source: Mapped[str]
+    char_count: Mapped[int]
+    sentence_count: Mapped[int]
+
+
+class ParseJob(Base):
+    __tablename__ = "parse_job"
+    id: Mapped[PK]
+    tenant_id: Mapped[UuidCol]
+    project_id: Mapped[UuidCol]
+    document_id: Mapped[UuidCol]
+    attempt_no: Mapped[int]
+    status: Mapped[str] = mapped_column(server_default=FetchedValue())
+    stage: Mapped[str | None]
+    checkpoints: Mapped[Json] = mapped_column(server_default=FetchedValue())
+    progress: Mapped[int] = mapped_column(server_default=FetchedValue())
+    cancel_requested_at: Mapped[Ts | None]
+    error_class: Mapped[str | None]
+    error_code: Mapped[str | None]
+    error_detail: Mapped[str | None]
+    prompt_versions: Mapped[Json | None]
+    quota_hold_id: Mapped[UuidCol | None]
+    queue_job_id: Mapped[int | None]
+    created_by: Mapped[UuidCol]
+    created_at: Mapped[DbTs]
+    started_at: Mapped[Ts | None]
+    finished_at: Mapped[Ts | None]
+
+
+class Citation(Base):
+    """出处：不可变（data-model §5.1）。"""
+
+    __tablename__ = "citation"
+    id: Mapped[PK]
+    tenant_id: Mapped[UuidCol]
+    project_id: Mapped[UuidCol]
+    document_id: Mapped[UuidCol]
+    normalized_sha256: Mapped[Bytes]
+    page_no: Mapped[int]
+    sentence_keys: Mapped[TextArray]
+    char_start: Mapped[int | None]
+    char_end: Mapped[int | None]
+    quads: Mapped[Json]
+    quote: Mapped[str]
+    created_by_kind: Mapped[str]
+    checks_passed: Mapped[TextArray]

server/src/bid_doctor/document/pipeline/convert.py

@@ -0,0 +1,60 @@
+"""转换：Word → 规范化 PDF（架构 §3.5，技术验证 S1）。PDF 原样作为规范化 PDF。
+
+TODO(S1)：
+  - 转换镜像要带按宋体行高调整过的字体包（BD Song / BD Hei）和 fontconfig 替换规则，现在用的是 Gotenberg 默认字体，
+    分页偏差会偏大（S1：默认配置 +17%～+37%）；
+  - WPS 生成的旧版 DOC 要先转 DOCX、修正字体字符集再转 PDF，否则正文可能是空白（S1 §1.3）。现在直接转。
+"""
+
+from __future__ import annotations
+
+import io
+import re
+import zipfile
+
+import httpx
+
+from bid_doctor.core.config import get_settings
+from bid_doctor.core.errors import AppError
+
+CJK_LATIN = {"SimSun", "NSimSun", "SimHei", "FangSong", "KaiTi", "Microsoft YaHei", "DengXian"}
+TIMEOUT = httpx.Timeout(300, connect=10)  # 服务端 240s 先超时，错误由转换服务明确返回
+
+
+def fix_wps_charset(docx: bytes) -> bytes:
+    """WPS 把中文字体标成 SYMBOL(02) 字符集时改成 GB2312(86)，否则 LibreOffice 会用符号字体排版（S1）。"""
+    src = zipfile.ZipFile(io.BytesIO(docx))
+    if "word/fontTable.xml" not in src.namelist():
+        return docx
+
+    def repl(m: re.Match[str]) -> str:
+        name, body = m.group(1), m.group(2)
+        if (re.search(r"[一-鿿]", name) or name in CJK_LATIN) and 'w:charset w:val="02"' in body:
+            body = body.replace('w:charset w:val="02"', 'w:charset w:val="86"')
+        return f'<w:font w:name="{name}">{body}</w:font>'
+
+    out = io.BytesIO()
+    with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as dst:
+        for item in src.infolist():
+            data = src.read(item.filename)
+            if item.filename == "word/fontTable.xml":
+                data = re.sub(
+                    r'<w:font w:name="([^"]+)">(.*?)</w:font>', repl, data.decode("utf-8"), flags=re.S
+                ).encode()
+            dst.writestr(item, data)
+    return out.getvalue()
+
+
+async def to_pdf(data: bytes, fmt: str) -> bytes:
+    if fmt == "pdf":
+        return data
+    if fmt == "docx":
+        data = fix_wps_charset(data)
+    async with httpx.AsyncClient(base_url=get_settings().gotenberg_url, timeout=TIMEOUT) as c:
+        r = await c.post("/forms/libreoffice/convert", files={"files": (f"input.{fmt}", data)})
+    if r.status_code == 200:
+        return r.content
+    if r.status_code in (400, 422):  # 文件本身转不了（加密、损坏）：输入问题，不扣点
+        raise AppError("file_corrupted", "这份 Word 文件无法转换，可能已加密或损坏")
+    r.raise_for_status()
+    raise AssertionError("unreachable")

server/src/bid_doctor/document/pipeline/runner.py

@@ -0,0 +1,194 @@
+"""解析流水线（架构 §3.4、§3.5）：格式识别 → 转换 → 文字层 → OCR 补全 → 抽取 → 证据校验 → 清单合并。
+
+- 每个阶段结束写检查点；重跑时已完成的阶段直接跳过（规范化 PDF 只生成一次，本来也不能重写）。
+- 阶段之间检查取消标记。
+- 结局由错误类别决定扣不扣点：基础设施错误、输入问题、取消 → 退回；成功 → 结算（架构 §3.6 的错误分类表）。
+TODO(结构识别)：章节、表格、分包识别还没做，现在只抽实质性条款。TODO(S5)：OCR 还没接，没有文字层的页直接报「需要 OCR」。
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import UTC, datetime
+from typing import Any
+from uuid import UUID
+
+import httpx
+from sqlalchemy import delete, select
+
+from bid_doctor.ai.providers.base import TransientError
+from bid_doctor.billing import service as billing
+from bid_doctor.checklist import extract
+from bid_doctor.checklist import service as checklist
+from bid_doctor.core import events, keys
+from bid_doctor.core.crypto import open_object, seal_object, sha256
+from bid_doctor.core.db import TenantContext, tenant_tx
+from bid_doctor.core.errors import AppError
+from bid_doctor.core.ids import uuid7
+from bid_doctor.core.storage import storage
+from bid_doctor.document.models import DocumentPage, ParseJob, SourceDocument
+from bid_doctor.document.pipeline import convert, text_layer
+from bid_doctor.document.service import doc_prefix
+from bid_doctor.notification.models import Notification
+from bid_doctor.project import service as projects
+
+log = logging.getLogger(__name__)
+PROGRESS = {"detect": 5, "convert": 15, "text_layer": 40, "ocr": 45, "extract": 55, "verify": 90, "merge": 95}
+PROFILE = {"converter": "gotenberg-8", "fonts": "gotenberg-default", "text_layer": "v1", "sentence_split": "s3-v1"}
+
+
+class Cancelled(Exception):
+    pass
+
+
+async def run(tenant_id: UUID, job_id: UUID) -> None:
+    ctx = TenantContext(tenant_id=tenant_id, user_id=UUID(int=0))
+    async with tenant_tx(ctx) as s:
+        job = (await s.execute(select(ParseJob).where(ParseJob.id == job_id))).scalar_one()
+        if job.status not in ("queued", "running"):
+            return  # 重复派发：已经有结局了
+        ctx = TenantContext(tenant_id=tenant_id, user_id=job.created_by, actor_name="系统")
+        job.status, job.started_at = "running", job.started_at or datetime.now(UTC)
+    try:
+        result = await _stages(ctx, job_id)
+    except Cancelled:
+        await _finish(ctx, job_id, "cancelled", "cancelled", None)
+    except AppError as e:
+        await _finish(ctx, job_id, "failed", "input", e.code)
+    except (TransientError, httpx.HTTPError) as e:
+        log.warning("parse job %s infra error: %s", job_id, e)
+        code = "model_unavailable" if isinstance(e, TransientError) else "converter_unavailable"
+        await _finish(ctx, job_id, "failed", "infra", code)
+    except Exception:
+        log.exception("parse job %s crashed", job_id)
+        await _finish(ctx, job_id, "failed", "infra", "internal_error")
+    else:
+        await _finish(ctx, job_id, "succeeded", None, None, result)
+
+
+async def _stage(ctx: TenantContext, job_id: UUID, stage: str) -> dict[str, Any]:
+    """进入一个阶段：写进度、检查取消标记，返回已有的检查点。"""
+    async with tenant_tx(ctx) as s:
+        job = (await s.execute(select(ParseJob).where(ParseJob.id == job_id))).scalar_one()
+        if job.cancel_requested_at:
+            raise Cancelled
+        job.stage, job.progress = stage, PROGRESS[stage]
+        checkpoints = dict(job.checkpoints or {})
+    await events.publish(
+        ctx.tenant_id,
+        "job.progress",
+        project_id=str(job.project_id),
+        job={"kind": "parse_job", "id": str(job_id), "status": "running", "stage": stage, "progress": PROGRESS[stage]},
+    )
+    return checkpoints
+
+
+async def _stages(ctx: TenantContext, job_id: UUID) -> dict[str, int]:
+    async with tenant_tx(ctx) as s:
+        job = (await s.execute(select(ParseJob).where(ParseJob.id == job_id))).scalar_one()
+        doc = (await s.execute(select(SourceDocument).where(SourceDocument.id == job.document_id))).scalar_one()
+        dk = await keys.data_key(s, ctx, doc.project_id)
+
+    await _stage(ctx, job_id, "detect")  # 上传完成时已识别过格式；这里只核对原件没被替换
+    original = open_object(dk, storage().get(doc.original_key), doc.original_key)
+    if sha256(original) != doc.original_sha256:
+        raise AppError("file_corrupted")
+
+    await _stage(ctx, job_id, "convert")
+    if doc.normalized_key is None:
+        pdf = await convert.to_pdf(original, doc.detected_format)
+        key = f"{doc_prefix(doc)}/normalized.pdf"
+        storage().put(key, seal_object(dk, pdf, key))
+        async with tenant_tx(ctx) as s:
+            d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
+            d.normalized_key, d.normalized_sha256, d.status = key, sha256(pdf), "normalizing"
+            d.pipeline_profile = PROFILE
+        doc.normalized_key, doc.normalized_sha256 = key, sha256(pdf)
+    else:
+        pdf = open_object(dk, storage().get(doc.normalized_key), doc.normalized_key)
+
+    await _stage(ctx, job_id, "text_layer")
+    pages = text_layer.extract(str(doc.id), pdf)
+    text_key = f"{doc_prefix(doc)}/text.json"
+    storage().put(
+        text_key, seal_object(dk, json.dumps([p.to_json() for p in pages], ensure_ascii=False).encode(), text_key)
+    )
+    async with tenant_tx(ctx) as s:
+        await s.execute(delete(DocumentPage).where(DocumentPage.document_id == doc.id))  # 阶段幂等：先清残留
+        for p in pages:
+            s.add(
+                DocumentPage(
+                    tenant_id=ctx.tenant_id,
+                    document_id=doc.id,
+                    page_no=p.page_no,
+                    width_pt=p.width,
+                    height_pt=p.height,
+                    rotation=p.rotation,
+                    text_source=p.text_source,
+                    char_count=p.char_count,
+                    sentence_count=len(p.sentences),
+                )
+            )
+        d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
+        d.text_layer_key, d.page_count, d.status = text_key, len(pages), "ready"
+        _, plan = await billing.current_plan(s, ctx)
+    if len(pages) > plan.limits["max_pages"]:
+        raise AppError("plan_limit_exceeded")
+
+    await _stage(ctx, job_id, "ocr")
+    empty = sum(p.text_source == "empty" for p in pages)
+    if pages and empty / len(pages) > 0.5:
+        raise AppError("ocr_required")
+
+    await _stage(ctx, job_id, "extract")
+    sentences = [x for p in pages for x in p.sentences]
+    extracted, notes = await extract.run(ctx, job_id, sentences)
+
+    await _stage(ctx, job_id, "verify")  # 三道校验在抽取时逐条做完；这里只是进度节点
+    await _stage(ctx, job_id, "merge")
+    async with tenant_tx(ctx) as s:
+        d = (await s.execute(select(SourceDocument).where(SourceDocument.id == doc.id))).scalar_one()
+        stats = await checklist.merge(s, ctx, d, job_id, extracted, notes)
+    return {**stats, "lots_found": 0}
+
+
+async def _finish(
+    ctx: TenantContext,
+    job_id: UUID,
+    status: str,
+    error_class: str | None,
+    code: str | None,
+    result: dict[str, int] | None = None,
+) -> None:
+    async with tenant_tx(ctx) as s:
+        job = (await s.execute(select(ParseJob).where(ParseJob.id == job_id))).scalar_one()
+        job.status, job.finished_at, job.error_class, job.error_code = status, datetime.now(UTC), error_class, code
+        job.progress = 100 if status == "succeeded" else job.progress
+        if result is not None:
+            job.checkpoints = {**(job.checkpoints or {}), "result": result}
+        if job.quota_hold_id:
+            if status == "succeeded":
+                await billing.settle(s, ctx, job.quota_hold_id)
+            else:
+                await billing.release(s, ctx, job.quota_hold_id)  # 失败、超时、取消都不扣点（红线 9）
+        if status == "succeeded":
+            await projects.mark_drafting(s, await projects.get(s, ctx, job.project_id))
+        if status != "cancelled":
+            s.add(
+                Notification(
+                    id=uuid7(),
+                    tenant_id=ctx.tenant_id,
+                    user_id=job.created_by,
+                    kind="parse_succeeded" if status == "succeeded" else "parse_failed",
+                    project_id=job.project_id,
+                    payload={"job_id": str(job.id)},
+                )
+            )
+        project_id = job.project_id
+    await events.publish(
+        ctx.tenant_id,
+        "job.finished",
+        project_id=str(project_id),
+        job={"kind": "parse_job", "id": str(job_id), "status": status, "progress": 100},
+    )

server/src/bid_doctor/document/pipeline/text_layer.py

@@ -0,0 +1,225 @@
+"""文字层：从规范化 PDF 取逐字坐标和格式 → 切句 → 每句生成 quads 和稳定编号（架构 §3.5，技术验证 S2、S3）。
+
+切句规则：
+  1. 同一行里字间空隙大于 1.5 个字号时断开（表格单元格、制表位），得到「行片段」；
+  2. 句子在 。！？； 处结束；行片段没写到右边距附近（段落末行）时也结束；
+  3. 句子不跨页。编号 <文档>:p<页>:s<页内序号>；规范化 PDF 不再重新生成，所以编号稳定。
+格式：斜体看字体名或字形倾斜；下划线看字下方的横线；加粗看字体名（S2：斜体且下划线按字符 100% 一致）。
+TODO(S3 改进)：表格按单元格切句；页眉页脚带单独识别（现在都算正文）。
+"""
+
+from __future__ import annotations
+
+import io
+import itertools
+import logging
+import re
+import unicodedata
+from dataclasses import dataclass, field
+from typing import Any
+
+import pdfplumber
+
+END = set("。！？；")
+WS = re.compile(r"[\s　]")
+# pdfminer 对缺 FontBBox 的字体逐字报警告，没有信息量
+logging.getLogger("pdfminer").setLevel(logging.ERROR)
+
+
+# 部首补充区没有 NFKC 映射，简化字形的部首要单独查表
+RADICAL_SUPPLEMENT = dict(
+    zip(
+        "⺟⺠⻅⻆⻉⻋⻓⻔⻘⻚⻛⻜⻝⻢⻣⻤⻥⻦⻧⻨⻩⻬⻮⻰⻳",
+        "母民见角贝车长门青页风飞食马骨鬼鱼鸟卤麦黄齐齿龙龟",
+        strict=True,
+    )
+)
+
+
+def _char(t: str) -> str:
+    """有的 PDF（如 Chromium 生成的）把汉字映射成康熙部首或部首补充区的码位（「目」成了 U+2F6C）。
+    只把这两个区换回常用汉字；全角标点等其他字符保持原样，否则引文和原文对不上。"""
+    if len(t) != 1:
+        return t
+    if "⼀" <= t <= "⿟":
+        return unicodedata.normalize("NFKC", t)
+    return RADICAL_SUPPLEMENT.get(t, t)
+
+
+@dataclass(slots=True)
+class Sentence:
+    key: str
+    page_no: int
+    seq: int
+    text: str
+    quads: list[list[float]]
+    fmt_runs: list[tuple[int, int, str]] = field(default_factory=list)  # [起, 止, 格式]，格式取 b / i / u 的组合
+    location: str = "body"
+
+    @property
+    def marked(self) -> str:
+        """给模型看的文本：斜体且下划线 → <iu>，只下划线 → <u>，只斜体 → <i>（S4 的输入格式）。"""
+        out, pos = [], 0
+        for start, end, f in self.fmt_runs:
+            tag = "iu" if "i" in f and "u" in f else "u" if "u" in f else "i" if "i" in f else ""
+            if not tag:
+                continue
+            out.append(self.text[pos:start])
+            out.append(f"<{tag}>{self.text[start:end]}</{tag}>")
+            pos = end
+        out.append(self.text[pos:])
+        return "".join(out)
+
+    def to_json(self) -> dict[str, Any]:
+        return {
+            "key": self.key,
+            "seq": self.seq,
+            "text": self.text,
+            "quads": self.quads,
+            "format_runs": [list(r) for r in self.fmt_runs],
+            "location": self.location,
+        }
+
+    @classmethod
+    def from_json(cls, page_no: int, d: dict[str, Any]) -> Sentence:
+        return cls(
+            key=d["key"],
+            page_no=page_no,
+            seq=d["seq"],
+            text=d["text"],
+            quads=d["quads"],
+            fmt_runs=[(a, b, f) for a, b, f in d["format_runs"]],
+            location=d["location"],
+        )
+
+
+@dataclass(slots=True)
+class Page:
+    page_no: int
+    width: float
+    height: float
+    rotation: int
+    char_count: int
+    sentences: list[Sentence]
+
+    @property
+    def text_source(self) -> str:
+        return "text" if self.char_count >= 20 else "empty"
+
+    def to_json(self) -> dict[str, Any]:
+        return {
+            "page_no": self.page_no,
+            "width_pt": self.width,
+            "height_pt": self.height,
+            "rotation": self.rotation,
+            "char_count": self.char_count,
+            "sentences": [s.to_json() for s in self.sentences],
+        }
+
+    @classmethod
+    def from_json(cls, d: dict[str, Any]) -> Page:
+        return cls(
+            page_no=d["page_no"],
+            width=d["width_pt"],
+            height=d["height_pt"],
+            rotation=d["rotation"],
+            char_count=d["char_count"],
+            sentences=[Sentence.from_json(d["page_no"], s) for s in d["sentences"]],
+        )
+
+
+def _flags(c: dict[str, Any], hlines: list[tuple[float, float, float]]) -> str:
+    name = str(c["fontname"]).split("+")[-1]
+    italic = "Italic" in name or "Oblique" in name or abs(c["matrix"][2]) > 0.1
+    mid = (c["x0"] + c["x1"]) / 2
+    # 下划线一般在基线下 1～2pt；字体下行部分小时会落在字符框内，所以窗口向上放宽 0.25 个字号
+    under = any(x0 - 0.5 <= mid <= x1 + 0.5 and -0.25 * c["size"] <= y - c["bottom"] <= 4 for x0, x1, y in hlines)
+    return ("b" if "Bold" in name else "") + ("i" if italic else "") + ("u" if under else "")
+
+
+def _fragments(chars: list[dict[str, Any]]) -> list[list[dict[str, Any]]]:
+    chars = sorted(chars, key=lambda c: (round(c["bottom"]), c["x0"]))
+    lines: list[list[dict[str, Any]]] = []
+    cur: list[dict[str, Any]] = []
+    base: float | None = None
+    for c in chars:
+        if base is not None and abs(c["bottom"] - base) > 0.45 * c["size"]:
+            lines.append(cur)
+            cur = []
+        if not cur:
+            base = c["bottom"]
+        cur.append(c)
+    if cur:
+        lines.append(cur)
+    frags: list[list[dict[str, Any]]] = []
+    for ln in lines:
+        ln.sort(key=lambda c: c["x0"])
+        f = [ln[0]]
+        for a, b in itertools.pairwise(ln):
+            if b["x0"] - a["x1"] > 1.5 * a["size"]:
+                frags.append(f)
+                f = []
+            f.append(b)
+        frags.append(f)
+    return frags
+
+
+def _page(doc_id: str, page: Any) -> Page:
+    hlines = [(ln["x0"], ln["x1"], ln["top"]) for ln in page.lines if abs(ln["top"] - ln["bottom"]) < 0.5]
+    hlines += [(r["x0"], r["x1"], r["top"]) for r in page.rects if r["height"] < 1.5 and r["width"] > 2]
+    chars = [c for c in page.chars if str(c["text"]).strip()]
+    right = max((c["x1"] for c in chars), default=page.width)
+    groups: list[list[tuple[int, dict[str, Any]]]] = []
+    cur: list[tuple[int, dict[str, Any]]] = []
+    for fi, frag in enumerate(_fragments(chars)):
+        for c in frag:
+            cur.append((fi, c))
+            if c["text"] in END:
+                groups.append(cur)
+                cur = []
+        if cur and frag[-1]["x1"] < right - 2 * frag[-1]["size"]:
+            groups.append(cur)
+            cur = []
+    if cur:
+        groups.append(cur)
+    sentences: list[Sentence] = []
+    for seq, g in enumerate(groups, 1):
+        quads: dict[int, list[float]] = {}
+        runs: list[tuple[int, int, str]] = []
+        text_parts: list[str] = []
+        for pos, (fi, c) in enumerate(g):
+            q = quads.setdefault(fi, [c["x0"], c["top"], c["x1"], c["bottom"]])
+            q[0], q[1], q[2], q[3] = min(q[0], c["x0"]), min(q[1], c["top"]), max(q[2], c["x1"]), max(q[3], c["bottom"])
+            text_parts.append(_char(str(c["text"])))
+            f = _flags(c, hlines)
+            if f and runs and runs[-1][1] == pos and runs[-1][2] == f:
+                runs[-1] = (runs[-1][0], pos + 1, f)
+            elif f:
+                runs.append((pos, pos + 1, f))
+        sentences.append(
+            Sentence(
+                key=f"{doc_id}:p{page.page_number}:s{seq}",
+                page_no=page.page_number,
+                seq=seq,
+                text="".join(text_parts),
+                quads=[[round(v, 2) for v in q] for q in quads.values()],
+                fmt_runs=runs,
+            )
+        )
+    return Page(
+        page_no=page.page_number,
+        width=float(page.width),
+        height=float(page.height),
+        rotation=int(page.rotation or 0) % 360,
+        char_count=len(chars),
+        sentences=sentences,
+    )
+
+
+def extract(doc_id: str, pdf: bytes) -> list[Page]:
+    with pdfplumber.open(io.BytesIO(pdf)) as doc:
+        return [_page(doc_id, p) for p in doc.pages]
+
+
+def norm(s: str) -> str:
+    return WS.sub("", s)

server/src/bid_doctor/document/service.py

@@ -0,0 +1,229 @@
+"""文档、解析任务（FR-M2-01、02，FR-M3-01）。"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+from uuid import UUID
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from bid_doctor.audit import service as audit
+from bid_doctor.billing import service as billing
+from bid_doctor.core import keys, queue
+from bid_doctor.core.crypto import seal_object, sha256
+from bid_doctor.core.db import TenantContext
+from bid_doctor.core.errors import AppError
+from bid_doctor.core.ids import uuid7
+from bid_doctor.core.storage import project_prefix, storage
+from bid_doctor.document import uploads
+from bid_doctor.document.citations import document_label
+from bid_doctor.document.models import ParseJob, SourceDocument
+from bid_doctor.project import service as projects
+
+PARSABLE = {"tender", "addendum", "clarification"}
+PARSE_TASK = "parse_document"
+
+
+def doc_prefix(doc: SourceDocument) -> str:
+    return f"{project_prefix(doc.tenant_id, doc.project_id)}/doc/{doc.id}"
+
+
+async def create(
+    s: AsyncSession, ctx: TenantContext, project_id: UUID, upload_id: UUID, role: str, volume_name: str | None
+) -> SourceDocument:
+    p = await projects.get(s, ctx, project_id)
+    projects.ensure_editable(p)
+    meta, data = await uploads.take(ctx, upload_id)
+    assert meta.detected_format is not None
+    seq = (
+        await s.execute(
+            select(func.coalesce(func.max(SourceDocument.seq_no), -1) + 1).where(
+                SourceDocument.project_id == project_id, SourceDocument.role == role
+            )
+        )
+    ).scalar_one()
+    if role == "tender":
+        seq = 0
+    doc_id = uuid7()
+    doc = SourceDocument(
+        id=doc_id,
+        tenant_id=ctx.tenant_id,
+        project_id=project_id,
+        role=role,
+        seq_no=seq,
+        volume_name=volume_name,
+        lineage_id=doc_id,
+        original_filename=meta.filename,
+        detected_format=meta.detected_format,
+        size_bytes=meta.size_bytes,
+        original_key="",
+        original_sha256=sha256(data),
+        page_count=meta.page_count,
+        uploaded_by=ctx.user_id,
+    )
+    doc.original_key = f"{doc_prefix(doc)}/original"
+    dk = await keys.data_key(s, ctx, project_id)
+    storage().put(doc.original_key, seal_object(dk, data, doc.original_key))
+    s.add(doc)
+    await audit.record(
+        s,
+        ctx,
+        "document.create",
+        "source_document",
+        doc.id,
+        project_id=project_id,
+        after={"role": role, "filename": meta.filename},
+    )
+    await s.flush()
+    await s.refresh(doc)
+    await uploads.discard(ctx, upload_id)
+    return doc
+
+
+async def get(s: AsyncSession, document_id: UUID) -> SourceDocument:
+    doc = (await s.execute(select(SourceDocument).where(SourceDocument.id == document_id))).scalar_one_or_none()
+    if doc is None:
+        raise AppError("not_found")
+    return doc
+
+
+async def start_parse(s: AsyncSession, ctx: TenantContext, document_id: UUID, expected_charge: int | None) -> ParseJob:
+    """预扣 1 次解析 → 建任务 → 入队，在同一个事务里（架构 §3.4）。"""
+    doc = await get(s, document_id)
+    if doc.role not in PARSABLE:
+        raise AppError("document_role_not_parsable")
+    projects.ensure_editable(await projects.get(s, ctx, doc.project_id))
+    if expected_charge is not None and expected_charge != 1:
+        raise AppError("billing_changed", charge=1)  # 用户看到的扣点和实际不一致：拒绝，不扣点（API4）
+    running = (
+        await s.execute(
+            select(ParseJob.id).where(ParseJob.document_id == doc.id, ParseJob.status.in_(("queued", "running")))
+        )
+    ).scalar_one_or_none()
+    if running:
+        raise AppError("job_already_running", job_id=str(running))
+    attempt = (
+        await s.execute(
+            select(func.coalesce(func.max(ParseJob.attempt_no), 0) + 1).where(ParseJob.document_id == doc.id)
+        )
+    ).scalar_one()
+    job = ParseJob(
+        id=uuid7(),
+        tenant_id=ctx.tenant_id,
+        project_id=doc.project_id,
+        document_id=doc.id,
+        attempt_no=attempt,
+        created_by=ctx.user_id,
+    )
+    s.add(job)
+    await s.flush()
+    job.quota_hold_id = await billing.hold(
+        s, ctx, unit="parse", ref_type="parse_job", ref_id=job.id, project_id=doc.project_id
+    )
+    job.queue_job_id = await queue.enqueue(
+        s, PARSE_TASK, {"tenant_id": str(ctx.tenant_id), "job_id": str(job.id)}, lock=f"doc:{doc.id}"
+    )
+    await audit.record(s, ctx, "parse_job.start", "parse_job", job.id, project_id=doc.project_id)
+    await s.flush()
+    await s.refresh(job)
+    return job
+
+
+async def request_cancel(s: AsyncSession, job_id: UUID) -> ParseJob:
+    job = await get_job(s, job_id)
+    if job.status not in ("queued", "running"):
+        raise AppError("job_finished")
+    job.cancel_requested_at = datetime.now(UTC)
+    await s.flush()
+    return job
+
+
+async def get_job(s: AsyncSession, job_id: UUID) -> ParseJob:
+    job = (await s.execute(select(ParseJob).where(ParseJob.id == job_id))).scalar_one_or_none()
+    if job is None:
+        raise AppError("not_found")
+    return job
+
+
+ERROR_MESSAGES = {
+    "file_corrupted": "这份文件无法转换，可能已加密或损坏",
+    "file_encrypted": "文件已加密，请上传未加密的版本",
+    "ocr_required": "这份文件大部分页面没有文字层（可能是扫描件），扫描件识别还没有上线",
+    "plan_limit_exceeded": "页数超出套餐上限",
+    "converter_unavailable": "文档转换服务暂时不可用",
+    "model_unavailable": "模型服务暂时不可用",
+    "internal_error": "系统出错了",
+}
+
+
+def job_to_api(job: ParseJob) -> dict[str, Any]:
+    error = None
+    if job.status in ("failed", "cancelled") and job.error_code:
+        error = {
+            "code": job.error_code,
+            "message": ERROR_MESSAGES.get(job.error_code, job.error_code),
+            "refunded": True,
+        }
+    return {
+        "id": job.id,
+        "document_id": job.document_id,
+        "attempt_no": job.attempt_no,
+        "status": job.status,
+        "stage": job.stage,
+        "progress": job.progress,
+        "cancel_requested": job.cancel_requested_at is not None,
+        "error": error,
+        "result": (job.checkpoints or {}).get("result"),
+        "created_at": job.created_at,
+        "finished_at": job.finished_at,
+    }
+
+
+async def doc_to_api(s: AsyncSession, doc: SourceDocument) -> dict[str, Any]:
+    latest = (
+        await s.execute(
+            select(ParseJob).where(ParseJob.document_id == doc.id).order_by(ParseJob.attempt_no.desc()).limit(1)
+        )
+    ).scalar_one_or_none()
+    return {
+        "id": doc.id,
+        "role": doc.role,
+        "label": document_label(doc.role, doc.seq_no),
+        "seq_no": doc.seq_no,
+        "volume_name": doc.volume_name,
+        "lineage_id": doc.lineage_id,
+        "is_latest_version": True,
+        "original_filename": doc.original_filename,
+        "detected_format": doc.detected_format,
+        "size_bytes": doc.size_bytes,
+        "page_count": doc.page_count,
+        "normalized_sha256": doc.normalized_sha256.hex() if doc.normalized_sha256 else None,
+        "status": doc.status,
+        "latest_parse_job": job_to_api(latest) if latest else None,
+        "uploaded_by": {"id": doc.uploaded_by, "name": ""},
+        "created_at": doc.created_at,
+    }
+
+
+async def summaries(s: AsyncSession, project_ids: list[UUID]) -> dict[UUID, dict[str, Any]]:
+    """工作台用：每个项目的招标文件数、正在跑的解析。"""
+    out: dict[UUID, dict[str, Any]] = {pid: {"tenders": 0, "job": None} for pid in project_ids}
+    if not project_ids:
+        return out
+    for pid, n in (
+        await s.execute(
+            select(SourceDocument.project_id, func.count())
+            .where(SourceDocument.project_id.in_(project_ids), SourceDocument.role == "tender")
+            .group_by(SourceDocument.project_id)
+        )
+    ).tuples():
+        out[pid]["tenders"] = n
+    for job in (
+        await s.execute(
+            select(ParseJob).where(ParseJob.project_id.in_(project_ids), ParseJob.status.in_(("queued", "running")))
+        )
+    ).scalars():
+        out[job.project_id]["job"] = {"kind": "parse_job", "id": job.id, "progress": job.progress, "label": "解析中"}
+    return out

server/src/bid_doctor/document/uploads.py

@@ -0,0 +1,162 @@
+"""分片上传、断点续传（架构 §5.2 第 7 条，api-design §4.3）。
+
+上传会话的元数据放在 Valkey（24 小时过期，丢了就重新上传）；分片按企业级数据密钥加密后写进对象存储。
+TODO(大文件)：现在合并时整份读进内存；200MB 以上改成流式合并、分段加密。
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import asdict, dataclass
+from datetime import UTC, datetime, timedelta
+from typing import Any
+from uuid import UUID
+
+from bid_doctor.billing import service as billing
+from bid_doctor.core import keys
+from bid_doctor.core.config import get_settings
+from bid_doctor.core.crypto import open_object, seal_object
+from bid_doctor.core.db import TenantContext, tenant_tx
+from bid_doctor.core.errors import AppError
+from bid_doctor.core.ids import uuid7
+from bid_doctor.core.storage import storage, upload_prefix
+from bid_doctor.core.valkey import valkey
+from bid_doctor.document.formats import detect
+
+TTL_S = 24 * 3600
+
+
+@dataclass(slots=True)
+class UploadMeta:
+    id: str
+    filename: str
+    size_bytes: int
+    purpose: str
+    part_size: int
+    part_count: int
+    sha256: str | None = None
+    detected_format: str | None = None
+    page_count: int | None = None
+    extension_mismatch: bool = False
+    completed_sha256: str | None = None
+
+
+def _key(ctx: TenantContext, upload_id: UUID | str) -> str:
+    return f"up:{ctx.tenant_id}:{upload_id}"
+
+
+async def _limits(ctx: TenantContext) -> dict[str, Any]:
+    async with tenant_tx(ctx) as s:
+        _, plan = await billing.current_plan(s, ctx)
+    return plan.limits
+
+
+async def create(
+    ctx: TenantContext, filename: str, size_bytes: int, purpose: str, sha256: str | None
+) -> dict[str, Any]:
+    limits = await _limits(ctx)
+    if size_bytes > limits["max_file_bytes"]:
+        raise AppError("plan_limit_exceeded", f"单个文件最大 {limits['max_file_bytes'] // (1024 * 1024)}MB")
+    part = get_settings().upload_part_size
+    meta = UploadMeta(
+        id=str(uuid7()),
+        filename=filename,
+        size_bytes=size_bytes,
+        purpose=purpose,
+        part_size=part,
+        part_count=max(1, -(-size_bytes // part)),
+        sha256=sha256,
+    )
+    await valkey().set(_key(ctx, meta.id), json.dumps(asdict(meta)), ex=TTL_S)
+    return await describe(ctx, meta.id)
+
+
+async def load(ctx: TenantContext, upload_id: UUID | str) -> UploadMeta:
+    raw = await valkey().get(_key(ctx, upload_id))
+    if raw is None:
+        raise AppError("not_found")
+    return UploadMeta(**json.loads(raw))
+
+
+async def describe(ctx: TenantContext, upload_id: UUID | str) -> dict[str, Any]:
+    meta = await load(ctx, upload_id)
+    parts = sorted(int(p) for p in await valkey().smembers(_key(ctx, upload_id) + ":parts"))
+    ttl = await valkey().ttl(_key(ctx, upload_id))
+    return {
+        "id": meta.id,
+        "part_size": meta.part_size,
+        "part_count": meta.part_count,
+        "received_parts": parts,
+        "expires_at": datetime.now(UTC) + timedelta(seconds=max(ttl, 0)),
+    }
+
+
+async def put_part(ctx: TenantContext, upload_id: UUID, part_no: int, data: bytes) -> None:
+    meta = await load(ctx, upload_id)
+    last = part_no == meta.part_count
+    expected = meta.size_bytes - meta.part_size * (meta.part_count - 1) if last else meta.part_size
+    if not 1 <= part_no <= meta.part_count or len(data) != expected:
+        raise AppError("validation_failed", f"第 {part_no} 片的大小应为 {expected} 字节")
+    async with tenant_tx(ctx) as s:
+        dk = await keys.data_key(s, ctx)
+    key = f"{upload_prefix(ctx.tenant_id, upload_id)}/part-{part_no:05d}"
+    storage().put(key, seal_object(dk, data, key))
+    await valkey().sadd(_key(ctx, upload_id) + ":parts", part_no)
+    await valkey().expire(_key(ctx, upload_id) + ":parts", TTL_S)
+
+
+async def complete(ctx: TenantContext, upload_id: UUID) -> dict[str, Any]:
+    meta = await load(ctx, upload_id)
+    got = {int(p) for p in await valkey().smembers(_key(ctx, upload_id) + ":parts")}
+    if got != set(range(1, meta.part_count + 1)):
+        raise AppError("upload_incomplete", received_parts=sorted(got))
+    async with tenant_tx(ctx) as s:
+        dk = await keys.data_key(s, ctx)
+    prefix = upload_prefix(ctx.tenant_id, upload_id)
+    chunks = [
+        open_object(dk, storage().get(k), k) for k in (f"{prefix}/part-{n:05d}" for n in range(1, meta.part_count + 1))
+    ]
+    data = b"".join(chunks)
+    digest = hashlib.sha256(data).hexdigest()
+    if meta.sha256 and meta.sha256 != digest:
+        raise AppError("upload_checksum_mismatch")
+    d = detect(data, meta.filename)  # 格式、加密、损坏都在这里拦下，不建任务、不扣点
+    limits = await _limits(ctx)
+    if d.page_count and d.page_count > limits["max_pages"]:
+        raise AppError("plan_limit_exceeded", f"单个文件最多 {limits['max_pages']} 页，这份有 {d.page_count} 页")
+    storage().put(f"{prefix}/file", seal_object(dk, data, f"{prefix}/file"))
+    for n in range(1, meta.part_count + 1):  # 分片用完即删
+        storage().delete_prefix(f"{prefix}/part-{n:05d}")
+    meta.detected_format, meta.page_count, meta.extension_mismatch, meta.completed_sha256 = (
+        d.format,
+        d.page_count,
+        d.extension_mismatch,
+        digest,
+    )
+    await valkey().set(_key(ctx, upload_id), json.dumps(asdict(meta)), ex=TTL_S)
+    return {
+        "id": meta.id,
+        "filename": meta.filename,
+        "size_bytes": meta.size_bytes,
+        "sha256": digest,
+        "detected_format": d.format,
+        "extension_mismatch": d.extension_mismatch,
+        "page_count": d.page_count,
+    }
+
+
+async def take(ctx: TenantContext, upload_id: UUID) -> tuple[UploadMeta, bytes]:
+    """取走已完成的上传（明文），挂到项目上以后删除暂存。"""
+    meta = await load(ctx, upload_id)
+    if not meta.completed_sha256:
+        raise AppError("upload_incomplete")
+    async with tenant_tx(ctx) as s:
+        dk = await keys.data_key(s, ctx)
+    key = f"{upload_prefix(ctx.tenant_id, upload_id)}/file"
+    return meta, open_object(dk, storage().get(key), key)
+
+
+async def discard(ctx: TenantContext, upload_id: UUID) -> None:
+    storage().delete_prefix(upload_prefix(ctx.tenant_id, upload_id))
+    await valkey().delete(_key(ctx, upload_id), _key(ctx, upload_id) + ":parts")