upload_dates 直接取现行配方（current_recipe_id）：发布、启用旧版本时它和当前快照、表结构一起换，不必再经快照和最后一期导入去找；补测试：修改配方把支付日期改成文字后只剩下单日期，回滚到旧版本两列又都是日期

backend/app/engine/dual_check_facts.py

@@ -309,27 +309,21 @@ def recipe_dates(recipe: Any) -> dict[str, list[str]]:
 async def upload_dates(session: Any, row: DataSource | None) -> dict[str, list[str]]:
     """上传表格建的数据源当前版本里，导入时认作日期的列：{表名: [列名]}（recipe_dates）。只读，不碰导入流程。
 
-    按当前快照找它的配方（source_versions.snapshot_recipe_id：快照记着按哪一版配方建的表，期 3 之前的快照取最后一期导入
-    的）：数据源的表结构（schema_cache）也出自当前快照，两者是同一个版本。读不出来的一律给空——手工登记的源、简单导入
-    （没有配方，回执里只有存储类型）、快照或配方记录不在了、配方解析不了：业务日期照旧不记，原因说按文字保存。"""
+    配方取现行的那一版（current_recipe_id）：发布、启用旧版本时它和当前快照、表结构（schema_cache）一起换
+    （table_versions.activate_snapshot），配方和表结构出自同一个版本。读不出来的一律给空——手工登记的源、简单导入
+    （没有配方，构建回执里只有存储类型）、配方记录不在了、配方解析不了：业务日期照旧不记，原因说按文字保存。"""
     from app.data.recipe_types import Recipe
-    from app.data.source_versions import snapshot_recipe_id
-    from app.db.models import SourceSnapshot, TableImport, TableRecipe
+    from app.db.models import TableRecipe
 
-    if row is None or (row.origin or "manual") != "upload" or not row.current_snapshot_id:
+    if row is None or (row.origin or "manual") != "upload" or not row.current_recipe_id:
         return {}
-    snap = await session.get(SourceSnapshot, row.current_snapshot_id)
-    if snap is None or snap.source_id != row.id:
-        return {}
-    last = await session.get(TableImport, str(snap.imports[-1])) if snap.imports else None
-    recipe_id = snapshot_recipe_id(snap, {last.id: last} if last is not None else {})
-    record = await session.get(TableRecipe, recipe_id) if recipe_id else None
+    record = await session.get(TableRecipe, row.current_recipe_id)
     if record is None or record.source_id != row.id:
         return {}
     try:
         return recipe_dates(Recipe.model_validate(record.recipe))
     except ValueError:
-        logger.warning("上传表格的配方记录解析不了，业务日期按文字保存处理：source=%s recipe=%s", row.id, recipe_id)
+        logger.warning("上传表格的配方记录解析不了，业务日期按文字保存处理：source=%s recipe=%s", row.id, record.id)
         return {}
 
 

backend/tests/test_dual_check_upload.py

@@ -92,13 +92,8 @@ def list_recipe(*, ordered: str = "DATE", paid: str = "DATE") -> dict:
             "tables": [{"name": TABLE, "grain": ["订单号"]}]}
 
 
-async def recipe_import(client, recipe: dict) -> tuple[str, str]:
-    """按配方导入：暂存 → 换成给定的配方 → 试运行 → 勾全部确认项提交。返回 (数据源名, id)。"""
-    name = f"dc_upload_{uuid.uuid4().hex[:8]}"
-    r = await client.post(f"{API}/imports/stage", files={"file": (f"{SHEET}_2026-09.xlsx", workbook(), XLSX)},
-                          data={"name": name})
-    assert r.status_code == 201, r.text
-    staging = r.json()["id"]
+async def commit_recipe(client, staging: str, recipe: dict) -> dict:
+    """暂存区换成给定的配方 → 试运行 → 勾全部确认项提交。返回提交的响应。"""
     r = await client.put(f"{API}/imports/{staging}/recipe", json={"recipe": recipe})
     assert r.status_code == 200 and r.json()["recipe_problems"] == [], r.text
     r = await client.post(f"{API}/imports/{staging}/trial", json={})
@@ -108,7 +103,17 @@ async def recipe_import(client, recipe: dict) -> tuple[str, str]:
     r = await client.post(f"{API}/imports/{staging}/commit", json={
         "trial_id": trial["trial_id"], "confirmations": [i["id"] for i in trial.get("confirm_items") or []]})
     assert r.status_code == 201, r.text
-    return name, r.json()["source"]["id"]
+    return r.json()
+
+
+async def recipe_import(client, recipe: dict) -> tuple[str, str]:
+    """按配方导入：暂存 → 换成给定的配方 → 试运行 → 提交。返回 (数据源名, id)。"""
+    name = f"dc_upload_{uuid.uuid4().hex[:8]}"
+    r = await client.post(f"{API}/imports/stage", files={"file": (f"{SHEET}_2026-09.xlsx", workbook(), XLSX)},
+                          data={"name": name})
+    assert r.status_code == 201, r.text
+    out = await commit_recipe(client, r.json()["id"], recipe)
+    return name, out["source"]["id"]
 
 
 async def simple_import(client) -> tuple[str, str]:
@@ -127,11 +132,11 @@ async def dates_of(source_id: str) -> dict:
 
 
 async def break_recipe(source_id: str) -> None:
-    """把数据源当前版本的配方记录改坏（造「配方读不出来」）。"""
+    """把数据源当前版本的配方记录改坏（造「配方读不出来」）。当前快照记的就是这一版。"""
     async with SessionLocal() as session:
         row = await session.get(DataSource, source_id)
-        snap = await session.get(SourceSnapshot, row.current_snapshot_id)
-        record = await session.get(TableRecipe, snap.recipe_id)
+        assert (await session.get(SourceSnapshot, row.current_snapshot_id)).recipe_id == row.current_recipe_id
+        record = await session.get(TableRecipe, row.current_recipe_id)
         record.recipe = {"recipe_format": "agentlab-recipe/0"}
         await session.commit()
 
@@ -193,6 +198,29 @@ async def test_an_upload_source_reads_the_date_columns_from_its_current_recipe(c
     assert await dates_of(paid_as_text[1]) == {TABLE: ["下单日期"]}
 
 
+async def current_snapshot(source_id: str) -> str:
+    async with SessionLocal() as session:
+        return (await session.get(DataSource, source_id)).current_snapshot_id
+
+
+async def test_the_dates_follow_the_current_version_of_the_source(client, dated):
+    """只认数据源当前版本的配方（表结构出自同一个版本）：修改配方把支付日期改成「文字」、提交之后只剩下单日期；回滚到
+    旧版本，两列又都是日期。"""
+    _, source_id = dated
+    first = await current_snapshot(source_id)
+    r = await client.post(f"{API}/{source_id}/redraft", json={})
+    assert r.status_code == 201, r.text
+    await commit_recipe(client, r.json()["id"], list_recipe(paid="TEXT"))
+    second = await current_snapshot(source_id)
+    assert second != first
+    assert await dates_of(source_id) == {TABLE: ["下单日期"]}
+    r = await client.post(f"{API}/{source_id}/snapshots/{first}/activate",
+                          json={"confirm": True, "expected_current_snapshot_id": second})
+    assert r.status_code == 200, r.text
+    assert await current_snapshot(source_id) == first
+    assert await dates_of(source_id) == {TABLE: ["下单日期", "支付日期"]}
+
+
 async def test_without_a_readable_recipe_no_column_is_taken_as_a_date(client, dated, tmp_path):
     """读不出配方的都不认：简单导入没有配方，配方记录坏了，手工登记的库，不存在的数据源。"""
     name, source_id = await simple_import(client)