From fcf4d8c948be2040570bef1fe7a92efc433956ce Mon Sep 17 00:00:00 2001
From: qiaoqichao <qiaoqichao@southportalhub.com>
Date: Tue, 29 Sep 2026 16:45:07 +0800
Subject: [PATCH] =?UTF-8?q?feat(v2):=20S1=20=E6=8E=A5=E5=8F=A3=E5=B1=82?=
 =?UTF-8?q?=E2=80=94=E2=80=94=E6=9D=90=E6=96=99=E5=AF=BC=E5=85=A5=E4=B8=8E?=
 =?UTF-8?q?=E4=BA=8B=E5=AE=9E=E5=86=B2=E7=AA=81=E7=9A=84=208=20=E4=B8=AA?=
 =?UTF-8?q?=E7=AB=AF=E7=82=B9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

T02 第三个子卡。命名空间按 T00/04 §1 定案 P1 挂 project-assistant 之下，
复用现有鉴权与同源代理，鉴权/代理代码零改动。

新增 backend/api/prescreen.py（7 路径 / 8 操作）+ test_prescreen_api.py（27 项）。
app.py 仅新增 1 个 include_router 与 1 个异常处理器注册。

落实：
- 404 统一口径（03 §2）：不存在与无访问权一律 404，不泄露对象存在性。
  既有 check_project 抛 403 且被 5 个模块共用，故不改它，由本模块 scope() 包装。
  实测：无权项目返回 404 NOT_FOUND 而非 403。
- 错误信封（D2）：只在新路由输出 {error:{code,message,retryable,request_id}}。
- E-09 与 E-05 精确区分：新增 store.IdempotencyConflict，不再靠字符串比对猜码。
- 客户端清单只接受 4 个字段；id/status/received_at 由服务端生成——
  状态机写权不下放。manifest 校验因此从"整对象"改为"4 字段 + 服务端补全"。

**实测发现的缺口并已修复**：Input(extra="forbid") 的校验错误由 FastAPI 校验层在
进入端点前抛出，不经过 run()，导致同一命名空间出现两种错误形状。现注册
RequestValidationError 处理器，但**只在 prescreen 路径生效**，其它路径交回
FastAPI 默认实现——既有端点形状经实测确认未变（含校验错误的形状）。

验证：
- scripts/verify.py 9/9 PASS（exit 0）；全量单测 225 项 OK（196 → 225）
- claude_use/T02/smoke_s1.py 在真实 app.py 上 10/10 PASS：覆盖路由注册、
  真实前缀下的 manager 鉴权放行、404 统一、信封形状、以及既有端点未被污染

Co-Authored-By: Claude Code <noreply@anthropic.com>
---
 backend/api/prescreen.py            | 307 ++++++++++++++++++++++++++
 backend/app.py                      |   9 +
 backend/services/assistant_store.py |  10 +-
 backend/services/material_import.py |  18 +-
 backend/test_prescreen_api.py       | 324 ++++++++++++++++++++++++++++
 backend/test_prescreen_storage.py   |  32 ++-
 6 files changed, 695 insertions(+), 5 deletions(-)
 create mode 100644 backend/api/prescreen.py
 create mode 100644 backend/test_prescreen_api.py

diff --git a/backend/api/prescreen.py b/backend/api/prescreen.py
new file mode 100644
index 0000000..59638de
--- /dev/null
+++ b/backend/api/prescreen.py
@@ -0,0 +1,307 @@
+"""S1 预审接口层：材料集合导入与事实确认。
+
+命名空间（claude_use/T00/04-first-slice-contracts.md §1 定案 P1）：
+    /api/xingbang/risk-agent/project-assistant/prescreen/...
+
+挂在 project-assistant 前缀之下，因此**复用现有鉴权与代理，零改动**：
+- `authorize()` 的角色→路径表已有 `path.startswith("project-assistant/")` → 客户经理放行；
+  reviewer 的写豁免同样覆盖该前缀（backend/api/risk_agent_access.py:49,53）。
+- 同源代理 `/api/risk-agent/[...path]` 已转发 Authorization / X-Xingbang-Demo-Role /
+  Idempotency-Key（src/app/api/risk-agent/[...path]/route.ts:7）。
+**不**使用设计草案原提的 `/api/prescreen` + `/api/core/prescreen`：那条路要求 admin
+角色，且需改三处鉴权/代理代码。
+
+错误信封（T00/04 §2 决策 D2）：**只在本模块内**输出
+`{error:{code,message,retryable,request_id}}`。不改 `api/project_assistant.translate()`，
+也不注册 app 级处理器——既有端点的 `{"detail": ...}` 形状完全不变。
+
+404 统一口径（03 §2）：不存在与无访问权一律对外 404，不泄露对象是否存在。
+既有 `check_project` 对无权项目抛 403 且被 5 个模块共用，故**不改它**，由本模块的
+`scope()` 把访问失败包装成 404；演示域策略仍按 403（E-03 语义）。
+"""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.exception_handlers import request_validation_exception_handler
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, ConfigDict, Field
+
+from api.project_assistant import enabled
+from api.risk_agent_access import authorize, check_project
+from contracts.prescreen import (
+    EVIDENCE_MAX,
+    MAX_BATCH_FILES,
+    NOTE_MAX,
+    ErrorCode,
+    S1_ERROR_HTTP,
+    S1_ERROR_RETRYABLE,
+)
+from services import assistant_store as store
+from services import fact_conflict, material_import
+from services.assistant_settings import flags
+from services.fact_conflict import FactConflictUnresolved
+from services.project_assistant import ProjectAssistantService
+
+router = APIRouter(
+    prefix="/api/xingbang/risk-agent/project-assistant/prescreen",
+    dependencies=[Depends(authorize), Depends(enabled)],
+    tags=["兴邦预审 S1"],
+)
+
+DEMO_PROJECTS = ("DEMO-A", "DEMO-B", "DEMO-C")
+NOT_FOUND_MESSAGE = "对象不可访问或不存在"
+
+
+# ---------------------------------------------------------------- 请求模型
+class Input(BaseModel):
+    """与既有 api/project_assistant.Input 同口径：禁止未知字段（E-07）。"""
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class ImportBatchInput(Input):
+    # 清单条目只有 4 个字段；id/status/received_at 由服务端生成（03 §9.4）。
+    # data_domain 不在请求里：从项目读取，避免客户端与项目数据域不一致。
+    manifest: list[dict] = Field(min_length=1, max_length=MAX_BATCH_FILES)
+
+
+class RetryInput(Input):
+    item_ids: list[str] = Field(min_length=1, max_length=MAX_BATCH_FILES)
+    expected_revision: int = Field(ge=1)
+
+
+class CancelInput(Input):
+    expected_revision: int = Field(ge=1)
+
+
+class ResolveInput(Input):
+    selected_evidence_ids: list[str] = Field(min_length=1, max_length=EVIDENCE_MAX)
+    resolution_note: str = Field(min_length=1, max_length=NOTE_MAX)
+    expected_revision: int = Field(ge=1)
+
+
+# ---------------------------------------------------------------- 错误信封
+def _envelope(request: Request, code: ErrorCode, message: str) -> JSONResponse:
+    request_id = request.headers.get("x-request-id") or store.uid("req")
+    return JSONResponse(
+        status_code=S1_ERROR_HTTP[code],
+        content={"error": {"code": code.value, "message": message,
+                           "retryable": S1_ERROR_RETRYABLE[code],
+                           "request_id": request_id}},
+    )
+
+
+def scope(request: Request, project_id: str):
+    """项目范围检查。通过返回 None；不通过返回 404/403 信封。
+
+    无权访问的项目按 404 处理（03 §2 不泄露对象存在性）；演示域被禁用是策略性
+    拒绝，按 403（E-03）。
+    """
+    try:
+        check_project(request, project_id)
+    except HTTPException:
+        return _envelope(request, ErrorCode.NOT_FOUND, NOT_FOUND_MESSAGE)
+    if project_id in DEMO_PROJECTS and not flags()["demo"]:
+        return _envelope(request, ErrorCode.ACCESS_DENIED, "演示数据域已禁用")
+    return None
+
+
+def run(request: Request, call, invalid_code: ErrorCode = ErrorCode.INVALID_REQUEST):
+    """执行并把异常映射为统一信封。
+
+    顺序重要：HTTPException 必须原样抛出，否则依赖注入阶段的 401/403 会被吞掉。
+    """
+    try:
+        return call()
+    except HTTPException:
+        raise
+    except FactConflictUnresolved as exc:
+        # E-17 的「确认动作被拒」面；运行级结果面是 HTTP 200 且不走此信封
+        return _envelope(request, ErrorCode.FACT_CONFLICT_UNRESOLVED, str(exc))
+    except store.IdempotencyConflict as exc:
+        return _envelope(request, ErrorCode.IDEMPOTENCY_CONFLICT, str(exc))
+    except store.Conflict as exc:
+        return _envelope(request, ErrorCode.REVISION_CONFLICT, str(exc))
+    except PermissionError as exc:
+        return _envelope(request, ErrorCode.ACCESS_DENIED, str(exc))
+    except KeyError:
+        return _envelope(request, ErrorCode.NOT_FOUND, NOT_FOUND_MESSAGE)
+    except ValueError as exc:
+        return _envelope(request, invalid_code, str(exc))
+
+
+# ---------------------------------------------------------------- 内部工具
+def _service() -> ProjectAssistantService:
+    return ProjectAssistantService()
+
+
+def _db():
+    return _service().connect()
+
+
+def _project(db, project_id: str) -> dict:
+    project = store.get(db, "project", project_id)  # 不存在 → KeyError → 404
+    if project.get("data_domain") not in ("demo", "production"):
+        raise ValueError("项目数据域异常，无法导入材料")
+    return project
+
+
+def _actor(request: Request) -> str:
+    return request.state.identity["id"]
+
+
+def _key(request: Request) -> str:
+    key = request.headers.get("idempotency-key", "")
+    if len(key) > 160:
+        raise ValueError("幂等键过长")
+    return key
+
+
+def _limit(value: int) -> int:
+    return max(1, min(value, 100))
+
+
+async def validation_handler(request: Request, exc: RequestValidationError):
+    """请求体/查询参数校验失败也走统一信封。
+
+    为什么需要它：`Input(extra="forbid")` 之类由 FastAPI 的校验层在**进入端点之前**
+    抛错，不经过本模块的 `run()`，若不处理就会在同一命名空间里出现两种错误形状
+    （信封 + FastAPI 默认的 `{"detail": ...}`）。这是 smoke_s1.py 实测发现的。
+
+    **非 prescreen 路径一律交回 FastAPI 默认处理器**，因此既有端点的错误形状
+    完全不变（有测试锁定这一隔离）。
+    """
+    if not request.url.path.startswith(router.prefix):
+        return await request_validation_exception_handler(request, exc)
+    errors = exc.errors() if callable(exc.errors) else exc.errors
+    first = errors[0] if errors else {}
+    where = ".".join(str(part) for part in first.get("loc", ()) if part != "body") or "请求"
+    return _envelope(request, ErrorCode.INVALID_REQUEST,
+                     f"{where}：{first.get('msg', '校验失败')}")
+
+
+# ---------------------------------------------------------------- 导入批次
+@router.post("/projects/{project_id}/import-batches", status_code=202)
+def create_import_batch(project_id: str, payload: ImportBatchInput, request: Request):
+    blocked = scope(request, project_id)
+    if blocked is not None:
+        return blocked
+
+    def action():
+        with _db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            project = _project(db, project_id)
+            batch = material_import.create(db, project_id, project["data_domain"],
+                                           payload.manifest, _actor(request), _key(request))
+        return {"batch": batch}
+
+    return run(request, action, ErrorCode.IMPORT_BATCH_INVALID)
+
+
+@router.get("/projects/{project_id}/import-batches")
+def list_import_batches(project_id: str, request: Request, limit: int = 20):
+    blocked = scope(request, project_id)
+    if blocked is not None:
+        return blocked
+
+    def action():
+        with _db() as db:
+            return material_import.listing(db, project_id, _limit(limit))
+
+    return run(request, action)
+
+
+@router.get("/import-batches/{batch_id}")
+def get_import_batch(batch_id: str, request: Request):
+    def action():
+        with _db() as db:
+            batch = material_import.load(db, batch_id)  # KeyError → 404
+        blocked = scope(request, batch["project_id"])
+        return blocked if blocked is not None else batch
+
+    return run(request, action)
+
+
+@router.post("/import-batches/{batch_id}/retry", status_code=202)
+def retry_import_batch(batch_id: str, payload: RetryInput, request: Request):
+    def action():
+        with _db() as db:
+            probe = material_import.load(db, batch_id)
+        blocked = scope(request, probe["project_id"])
+        if blocked is not None:
+            return blocked
+        with _db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            return material_import.retry(db, batch_id, payload.item_ids,
+                                         _actor(request), payload.expected_revision)
+
+    return run(request, action, ErrorCode.IMPORT_BATCH_INVALID)
+
+
+@router.post("/import-batches/{batch_id}/cancel")
+def cancel_import_batch(batch_id: str, payload: CancelInput, request: Request):
+    def action():
+        with _db() as db:
+            probe = material_import.load(db, batch_id)
+        blocked = scope(request, probe["project_id"])
+        if blocked is not None:
+            return blocked
+        with _db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            return material_import.cancel(db, batch_id, _actor(request), payload.expected_revision)
+
+    return run(request, action, ErrorCode.IMPORT_BATCH_INVALID)
+
+
+# ---------------------------------------------------------------- 事实冲突
+@router.get("/projects/{project_id}/fact-conflicts")
+def list_fact_conflicts(project_id: str, request: Request, status: str | None = None,
+                        limit: int = 20):
+    blocked = scope(request, project_id)
+    if blocked is not None:
+        return blocked
+    if status not in (None, "unresolved", "resolved"):
+        return _envelope(request, ErrorCode.INVALID_REQUEST,
+                         "状态取值须为 unresolved 或 resolved")
+
+    def action():
+        with _db() as db:
+            return fact_conflict.listing(db, project_id, status, _limit(limit))
+
+    return run(request, action)
+
+
+@router.get("/fact-conflicts/{conflict_id}")
+def get_fact_conflict(conflict_id: str, request: Request):
+    def action():
+        with _db() as db:
+            row = fact_conflict.load(db, conflict_id)
+        blocked = scope(request, row["project_id"])
+        return blocked if blocked is not None else row
+
+    return run(request, action)
+
+
+@router.post("/fact-conflicts/{conflict_id}/resolve")
+def resolve_fact_conflict(conflict_id: str, payload: ResolveInput, request: Request):
+    def action():
+        with _db() as db:
+            probe = fact_conflict.load(db, conflict_id)
+        blocked = scope(request, probe["project_id"])
+        if blocked is not None:
+            return blocked
+        with _db() as db:
+            db.execute("BEGIN IMMEDIATE")
+            project = _project(db, probe["project_id"])
+            # resolve 就地写 project 的 values/selected_sources/facts，因此必须保存它；
+            # 不要在此重复计算 facts 的列名——resolve 已按 FIELDS 反查过。
+            resolved = fact_conflict.resolve(db, conflict_id, project,
+                                             payload.selected_evidence_ids,
+                                             payload.resolution_note,
+                                             _actor(request), payload.expected_revision)
+            store.put(db, "project", project)
+        return resolved
+
+    return run(request, action)
diff --git a/backend/app.py b/backend/app.py
index 0c427fc..d25c288 100644
--- a/backend/app.py
+++ b/backend/app.py
@@ -95,6 +95,15 @@ from api.project_assistant import router as project_assistant_router
 from api.operations_center import router as operations_router
 app.include_router(operations_router)
 app.include_router(project_assistant_router)
+# S1 预审接口：必须注册在 project_assistant 之后（更具体的前缀先匹配；
+# 其路径为 .../project-assistant/prescreen/...）——见 claude_use/T00/04 §1
+from fastapi.exceptions import RequestValidationError
+from api.prescreen import router as prescreen_router, validation_handler
+
+# 只为 prescreen 路径输出统一错误信封；处理器内部对其它路径交回 FastAPI 默认实现，
+# 因此既有端点的 {"detail": ...} 形状不变（有测试与冒烟锁定该隔离）。
+app.add_exception_handler(RequestValidationError, validation_handler)
+app.include_router(prescreen_router)
 
 
 @app.get("/")
diff --git a/backend/services/assistant_store.py b/backend/services/assistant_store.py
index 83673fe..0163d0e 100644
--- a/backend/services/assistant_store.py
+++ b/backend/services/assistant_store.py
@@ -103,7 +103,7 @@ def idempotent_lookup(db, actor, operation, key, request_hash):
     if not prior:
         return None
     if prior[0] != request_hash:
-        raise Conflict("幂等键已用于不同请求")
+        raise IdempotencyConflict("幂等键已用于不同请求")
     return prior[1]
 
 
@@ -160,3 +160,11 @@ def time_listing(db, project_ids=None):
 
 class Conflict(ValueError):
     pass
+
+
+class IdempotencyConflict(Conflict):
+    """幂等键已用于不同载荷（E-09 IDEMPOTENCY_CONFLICT）。
+
+    与普通 Conflict（E-05 REVISION_CONFLICT）区分：两者的用户文案与恢复动作不同，
+    接口层需要能分别映射为不同的错误码，故不用字符串比对来猜。
+    """
diff --git a/backend/services/material_import.py b/backend/services/material_import.py
index 89087e7..51e2afd 100644
--- a/backend/services/material_import.py
+++ b/backend/services/material_import.py
@@ -27,6 +27,11 @@ PARSER_VERSION = "assistant_materials.parse@1"
 
 _STATUS_ORDER = ("queued", "parsing", "parsed", "failed")
 
+# 客户端在清单里**只能**声明这 4 个字段（docs/v2/03-detailed-design.md §9.4）。
+# id / status / received_at / document_id / error_code 一律由服务端决定——
+# 让客户端自报 id 或 status 等于把状态机的写权交给调用方。
+MANIFEST_FIELDS = frozenset({"relative_path", "mime_type", "size_bytes", "sha256"})
+
 
 def _derive_status(items: list[dict]) -> str:
     """由逐项状态派生批次状态。
@@ -57,8 +62,19 @@ def _validate_manifest(manifest: list[dict]) -> list[dict]:
     items: list[dict] = []
     seen_sha: set[str] = set()
     for index, entry in enumerate(manifest):
+        if not isinstance(entry, dict):
+            raise ValueError(f"第 {index + 1} 个文件不是对象")
+        extra = sorted(set(entry) - MANIFEST_FIELDS)
+        if extra:
+            raise ValueError(f"第 {index + 1} 个文件包含不允许由客户端声明的字段：{extra}")
+
+        # id / status / received_at 由服务端决定；observed_at 留空——
+        # 导入时间不冒充材料发生时间（14 §3），未知保持 null（BR-02）
+        draft = dict(entry, id=store.uid("item"), status="queued",
+                     received_at=store.now(), document_id=None,
+                     error_code=None, observed_at=None)
         try:
-            parsed = ImportItem(**entry)
+            parsed = ImportItem(**draft)
         except ValidationError as exc:
             # 逐项错误沿用 E-01/E-16 的文案口径，指明是第几项
             first = exc.errors()[0]
diff --git a/backend/test_prescreen_api.py b/backend/test_prescreen_api.py
new file mode 100644
index 0000000..c70fd59
--- /dev/null
+++ b/backend/test_prescreen_api.py
@@ -0,0 +1,324 @@
+"""S1 接口层测试：材料集合导入与事实确认的 HTTP 契约。
+
+只挂 prescreen 路由到独立 app，数据库指向临时文件——不触碰真实库。
+"""
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from fastapi import FastAPI
+from fastapi.exceptions import RequestValidationError
+from fastapi.testclient import TestClient
+
+from api import prescreen as api
+from services import assistant_store as store
+from services import fact_conflict
+from services.project_assistant import ProjectAssistantService
+
+HEADERS = {"X-Xingbang-Demo-Role": "manager"}
+# 演示身份可访问的项目：DEMO-A/B/C、LAB-001、XB-2026-001
+PROJECT = "DEMO-B"
+FOREIGN_PROJECT = "OTHER-9"  # 不在演示身份的项目列表里
+
+
+def manifest(name="a.csv", sha="a" * 64, size=100):
+    return {"relative_path": name, "sha256": sha, "size_bytes": size, "mime_type": "text/csv"}
+
+
+class PrescreenApiTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
+        self.path = Path(self.temp.name)
+        self.service = ProjectAssistantService(self.path / "assistant.sqlite3")
+        self.patches = [
+            patch.object(api, "_service", lambda: self.service),
+            patch.dict(os.environ, {"XINGBANG_AUTH_USERS": "", "XINGBANG_REQUIRE_AUTH": "0",
+                                    "XINGBANG_LOCAL_MODEL_URL": ""}),
+        ]
+        for p in self.patches:
+            p.start()
+        app = FastAPI()
+        # 必须与 app.py 的注册保持一致：否则本测试的 app 不忠实于生产接线，
+        # 校验错误会走 FastAPI 默认形状而测不到信封。
+        app.add_exception_handler(RequestValidationError, api.validation_handler)
+        app.include_router(api.router)
+        self.client = TestClient(app)
+        self.base = "/api/xingbang/risk-agent/project-assistant/prescreen"
+
+    def tearDown(self):
+        self.client.close()
+        for p in reversed(self.patches):
+            p.stop()
+        self.temp.cleanup()
+
+    # ------------------------------------------------------------ 工具
+    def post(self, path, body=None, headers=None):
+        return self.client.post(self.base + path, json=body if body is not None else {},
+                                headers={**HEADERS, **(headers or {})})
+
+    def get(self, path, headers=None):
+        return self.client.get(self.base + path, headers={**HEADERS, **(headers or {})})
+
+    def error(self, response):
+        body = response.json()
+        self.assertIn("error", body, body)
+        return body["error"]
+
+    def seed_conflict(self):
+        """直接经服务层造一个未解决冲突（本文件测的是 HTTP 层，不重复测存储层）。
+
+        注意：演示种子项目 DEMO-B 可能**已有**已确认的 facility.amount，
+        所以不能假设"首次确认一定无冲突"。这里按现状自适应。
+        传 unit=None 使候选值不做万元→元换算，与已确认值同口径可比。
+        """
+
+        def cand(cid, value, source, locator):
+            return {"id": cid, "field": "facility.amount", "value": value,
+                    "unit": None, "source": source, "locator": locator}
+
+        with self.service.connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            project = store.get(db, "project", PROJECT)
+            existing = project.get("values", {}).get("facility.amount")
+            if existing is None:
+                payload = fact_conflict.detect(project, cand("c0", "1000", "a.csv", "第2行"),
+                                               "demo-manager", "初始确认")
+                store.put(db, "project", project)
+                self.assertIsNone(payload, "首次确认不应产生冲突")
+                project = store.get(db, "project", PROJECT)
+                existing = project["values"]["facility.amount"]
+
+            different = next(v for v in ("777", "888", "999") if str(v) != str(existing))
+            payload = fact_conflict.detect(project, cand("c1", different, "b.csv", "第9行"),
+                                           "demo-manager", "第二来源")
+            store.put(db, "project", project)
+            self.assertIsNotNone(payload, "与已确认值不同，应产生冲突")
+            return fact_conflict.create(db, PROJECT, payload, "demo-manager")
+
+    # ------------------------------------------------------------ 导入批次
+    def test_create_batch_returns_202_queued(self):
+        response = self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [manifest()]})
+        self.assertEqual(response.status_code, 202, response.text)
+        batch = response.json()["batch"]
+        self.assertEqual(batch["status"], "queued")
+        self.assertEqual(batch["data_domain"], "demo")
+        self.assertEqual(batch["revision"], 1)
+        self.assertTrue(batch["items"][0]["id"].startswith("item-"))
+        self.assertEqual(batch["items"][0]["status"], "queued")
+        self.assertIsNone(batch["items"][0]["observed_at"])
+
+    def test_client_supplied_server_fields_rejected(self):
+        bad = dict(manifest(), status="parsed")
+        response = self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [bad]})
+        self.assertEqual(response.status_code, 422, response.text)
+        self.assertEqual(self.error(response)["code"], "IMPORT_BATCH_INVALID")
+
+    def test_absolute_path_rejected_with_envelope(self):
+        response = self.post(f"/projects/{PROJECT}/import-batches",
+                             {"manifest": [manifest(name="C:/x.csv")]})
+        self.assertEqual(response.status_code, 422)
+        error = self.error(response)
+        self.assertEqual(error["code"], "IMPORT_BATCH_INVALID")
+        self.assertFalse(error["retryable"])
+        self.assertTrue(error["request_id"])
+
+    def test_unknown_body_field_rejected_with_envelope(self):
+        """校验层在进入端点前抛错，也必须走统一信封（否则同命名空间两种形状）。"""
+        response = self.post(f"/projects/{PROJECT}/import-batches",
+                             {"manifest": [manifest()], "data_domain": "production"})
+        self.assertEqual(response.status_code, 422, response.text)
+        error = self.error(response)
+        self.assertEqual(error["code"], "INVALID_REQUEST")
+        self.assertTrue(error["request_id"])
+
+    def test_invalid_query_param_rejected_with_envelope(self):
+        response = self.get(f"/projects/{PROJECT}/import-batches?limit=notanumber")
+        self.assertEqual(response.status_code, 422, response.text)
+        self.assertEqual(self.error(response)["code"], "INVALID_REQUEST")
+
+    def test_empty_manifest_rejected(self):
+        response = self.post(f"/projects/{PROJECT}/import-batches", {"manifest": []})
+        self.assertEqual(response.status_code, 422, response.text)
+
+    def test_foreign_project_is_404_not_403(self):
+        # 03 §2：不存在与无访问权统一 404，不泄露对象是否存在
+        response = self.post(f"/projects/{FOREIGN_PROJECT}/import-batches",
+                             {"manifest": [manifest()]})
+        self.assertEqual(response.status_code, 404, response.text)
+        self.assertEqual(self.error(response)["code"], "NOT_FOUND")
+
+    def test_get_batch_by_id(self):
+        created = self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [manifest()]}).json()["batch"]
+        response = self.get(f"/import-batches/{created['id']}")
+        self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual(response.json()["id"], created["id"])
+
+    def test_get_unknown_batch_is_404_envelope(self):
+        response = self.get("/import-batches/import-does-not-exist")
+        self.assertEqual(response.status_code, 404)
+        error = self.error(response)
+        self.assertEqual(error["code"], "NOT_FOUND")
+        # 文案不暴露对象是否存在
+        self.assertEqual(error["message"], "对象不可访问或不存在")
+
+    def test_idempotent_replay_returns_same_batch(self):
+        body = {"manifest": [manifest()]}
+        first = self.post(f"/projects/{PROJECT}/import-batches", body,
+                          headers={"Idempotency-Key": "k1"}).json()["batch"]
+        again = self.post(f"/projects/{PROJECT}/import-batches", body,
+                          headers={"Idempotency-Key": "k1"}).json()["batch"]
+        self.assertEqual(first["id"], again["id"])
+
+    def test_idempotency_conflict_is_409(self):
+        self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [manifest()]},
+                  headers={"Idempotency-Key": "k2"})
+        response = self.post(f"/projects/{PROJECT}/import-batches",
+                             {"manifest": [manifest(sha="c" * 64)]},
+                             headers={"Idempotency-Key": "k2"})
+        self.assertEqual(response.status_code, 409, response.text)
+        self.assertEqual(self.error(response)["code"], "IDEMPOTENCY_CONFLICT")
+
+    def test_overlong_idempotency_key_rejected(self):
+        response = self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [manifest()]},
+                             headers={"Idempotency-Key": "x" * 200})
+        self.assertEqual(response.status_code, 422, response.text)
+
+    def test_list_batches_for_project(self):
+        self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [manifest()]})
+        response = self.get(f"/projects/{PROJECT}/import-batches")
+        self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual(len(response.json()["items"]), 1)
+
+    def test_retry_only_failed_items(self):
+        batch = self.post(f"/projects/{PROJECT}/import-batches",
+                          {"manifest": [manifest(name="ok.csv"),
+                                        manifest(name="bad.csv", sha="b" * 64)]}).json()["batch"]
+        ids = [item["id"] for item in batch["items"]]
+        with self.service.connect() as db:
+            from services import material_import
+            material_import.mark_item(db, batch["id"], ids[0], "parsed", document_id="doc-1")
+            batch = material_import.mark_item(db, batch["id"], ids[1], "failed",
+                                              error_code="MATERIAL_PARSE_FAILED")
+        response = self.post(f"/import-batches/{batch['id']}/retry",
+                            {"item_ids": [ids[1]], "expected_revision": batch["revision"]})
+        # 202 表示已接受重试；但该请求是同步执行的，直接看返回体
+        self.assertIn(response.status_code, (200, 202), response.text)
+        items = {item["id"]: item["status"] for item in response.json()["items"]}
+        self.assertEqual(items[ids[1]], "queued")
+
+    def test_retry_with_stale_revision_is_409(self):
+        batch = self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [manifest()]}).json()["batch"]
+        with self.service.connect() as db:
+            from services import material_import
+            batch = material_import.mark_item(db, batch["id"], batch["items"][0]["id"], "failed",
+                                              error_code="MATERIAL_PARSE_FAILED")
+        response = self.post(f"/import-batches/{batch['id']}/retry",
+                             {"item_ids": [batch["items"][0]["id"]], "expected_revision": 1})
+        self.assertEqual(response.status_code, 409, response.text)
+        self.assertEqual(self.error(response)["code"], "REVISION_CONFLICT")
+
+    def test_cancel_then_retry_refused(self):
+        batch = self.post(f"/projects/{PROJECT}/import-batches", {"manifest": [manifest()]}).json()["batch"]
+        cancelled = self.post(f"/import-batches/{batch['id']}/cancel",
+                              {"expected_revision": batch["revision"]})
+        self.assertEqual(cancelled.status_code, 200, cancelled.text)
+        self.assertEqual(cancelled.json()["status"], "cancelled")
+        again = self.post(f"/import-batches/{batch['id']}/cancel",
+                          {"expected_revision": cancelled.json()["revision"]})
+        self.assertEqual(again.status_code, 409, again.text)
+
+    # ------------------------------------------------------------ 事实冲突
+    def test_list_conflicts_and_filter(self):
+        conflict = self.seed_conflict()
+        response = self.get(f"/projects/{PROJECT}/fact-conflicts?status=unresolved")
+        self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual([row["id"] for row in response.json()["items"]], [conflict["id"]])
+        empty = self.get(f"/projects/{PROJECT}/fact-conflicts?status=resolved")
+        self.assertEqual(empty.json()["items"], [])
+
+    def test_invalid_status_filter_rejected(self):
+        response = self.get(f"/projects/{PROJECT}/fact-conflicts?status=whatever")
+        self.assertEqual(response.status_code, 422, response.text)
+        self.assertEqual(self.error(response)["code"], "INVALID_REQUEST")
+
+    def test_get_conflict_by_id(self):
+        conflict = self.seed_conflict()
+        response = self.get(f"/fact-conflicts/{conflict['id']}")
+        self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual(response.json()["status"], "unresolved")
+
+    def test_resolve_happy_path_adopts_value_and_persists(self):
+        conflict = self.seed_conflict()
+        chosen = conflict["candidates"][1]["evidence_id"]
+        response = self.post(f"/fact-conflicts/{conflict['id']}/resolve",
+                             {"selected_evidence_ids": [chosen],
+                              "resolution_note": "以合同金额为准",
+                              "expected_revision": conflict["revision"]})
+        self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual(response.json()["status"], "resolved")
+        # 值必须真的落到项目事实里（resolve 就地改 project，接口层负责保存）
+        with self.service.connect() as db:
+            project = store.get(db, "project", PROJECT)
+        self.assertEqual(str(project["values"]["facility.amount"]),
+                         str(conflict["candidates"][1]["value"]))
+        self.assertEqual(project["selected_sources"]["facility.amount"], chosen)
+        self.assertEqual(project["facts"]["facility_amount"],
+                         project["values"]["facility.amount"])
+
+    def test_resolve_with_foreign_evidence_is_422(self):
+        conflict = self.seed_conflict()
+        response = self.post(f"/fact-conflicts/{conflict['id']}/resolve",
+                             {"selected_evidence_ids": ["ev-nope"],
+                              "resolution_note": "说明",
+                              "expected_revision": conflict["revision"]})
+        self.assertEqual(response.status_code, 422, response.text)
+        self.assertEqual(self.error(response)["code"], "INVALID_REQUEST")
+
+    def test_resolve_twice_is_409(self):
+        conflict = self.seed_conflict()
+        chosen = conflict["candidates"][1]["evidence_id"]
+        self.post(f"/fact-conflicts/{conflict['id']}/resolve",
+                  {"selected_evidence_ids": [chosen], "resolution_note": "说明",
+                   "expected_revision": conflict["revision"]})
+        again = self.post(f"/fact-conflicts/{conflict['id']}/resolve",
+                          {"selected_evidence_ids": [chosen], "resolution_note": "再来",
+                           "expected_revision": conflict["revision"]})
+        self.assertEqual(again.status_code, 409, again.text)
+
+    def test_resolve_unknown_conflict_is_404(self):
+        response = self.post("/fact-conflicts/fc-nope/resolve",
+                             {"selected_evidence_ids": ["ev-1"], "resolution_note": "说明",
+                              "expected_revision": 1})
+        self.assertEqual(response.status_code, 404, response.text)
+
+    # ------------------------------------------------------------ 边界
+    def test_envelope_shape_is_stable(self):
+        response = self.get("/import-batches/nope")
+        error = self.error(response)
+        self.assertEqual(set(error), {"code", "message", "retryable", "request_id"})
+
+    def test_existing_endpoints_still_use_detail_shape(self):
+        """D2 决策的证据：本模块以外不产出 error 信封。"""
+        import api.project_assistant as legacy
+        self.assertFalse(hasattr(legacy, "_envelope"))
+        self.assertFalse(hasattr(legacy, "S1_ERROR_HTTP"))
+
+    def test_router_is_mounted_under_project_assistant_prefix(self):
+        # 命名空间定案 P1：零鉴权改动的关键前提是前缀在 project-assistant 之下
+        self.assertEqual(api.router.prefix,
+                         "/api/xingbang/risk-agent/project-assistant/prescreen")
+
+    def test_storage_layer_untouched_by_api_layer(self):
+        """接口层不直接写 SQL，只经服务层——避免出现第二份业务规则。"""
+        import api.prescreen as module
+        with open(module.__file__, encoding="utf-8") as handle:
+            source = handle.read()
+        for banned in ("INSERT INTO", "UPDATE ", "DELETE FROM", "SELECT "):
+            self.assertNotIn(banned, source)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/backend/test_prescreen_storage.py b/backend/test_prescreen_storage.py
index 4b6bce1..c1db381 100644
--- a/backend/test_prescreen_storage.py
+++ b/backend/test_prescreen_storage.py
@@ -19,9 +19,11 @@ from services.assistant_materials import FIELDS
 
 
 def manifest_entry(name="a.csv", sha="a" * 64, size=100, mime="text/csv"):
-    return {"id": store.uid("item"), "relative_path": name, "sha256": sha,
-            "size_bytes": size, "mime_type": mime, "status": "queued",
-            "received_at": store.now()}
+    """客户端清单条目：只有设计允许的 4 个字段（03 §9.4）。
+
+    id / status / received_at 由服务端生成——见 test_client_cannot_declare_server_fields。
+    """
+    return {"relative_path": name, "sha256": sha, "size_bytes": size, "mime_type": mime}
 
 
 class StoreTestCase(unittest.TestCase):
@@ -70,6 +72,30 @@ class ImportBatchTests(StoreTestCase):
             self.assertEqual(batch["revision"], 1)
             self.assertEqual(batch["parser_version"], material_import.PARSER_VERSION)
 
+    def test_client_cannot_declare_server_fields(self):
+        """状态机的写权不交给调用方：客户端不得自报 id / status / received_at。"""
+        for field, value in [("status", "parsed"), ("id", "item-x"),
+                             ("received_at", store.now()), ("document_id", "doc-1")]:
+            with self.subTest(field=field):
+                entry = manifest_entry()
+                entry[field] = value
+                with store.transaction(self.db_path) as db:
+                    with self.assertRaises(ValueError) as ctx:
+                        material_import.create(db, self.project_id, "demo", [entry], self.actor)
+                    self.assertIn("不允许由客户端声明", str(ctx.exception))
+
+    def test_server_generates_item_identity_and_queued_status(self):
+        with store.transaction(self.db_path) as db:
+            batch = material_import.create(db, self.project_id, "demo",
+                                           [manifest_entry()], self.actor)
+            item = batch["items"][0]
+            self.assertTrue(item["id"].startswith("item-"))
+            self.assertEqual(item["status"], "queued")
+            self.assertIsNotNone(item["received_at"])
+            # BR-02：发生时间未知就保持 null，不用导入时间冒充
+            self.assertIsNone(item["observed_at"])
+            self.assertIsNone(item["document_id"])
+
     def test_empty_manifest_rejected(self):
         with store.transaction(self.db_path) as db:
             with self.assertRaises(ValueError) as ctx: