起草按复核修正：读回复出错也记进用量；补问时列出了、字段留空不算没写出来；留空的列看目录里已有的；一项都没写出来的宽表逐段写范围；提醒照约定的说法

- 读回复时出了意料之外的错：这次调用照样记进用量（算一次出错），再交给 _guarded 记这一块没写出来；以前 _take 抛异常
  时这次调用不进 calls、token、failed_calls，也不进提醒的窗口。校验（catalog._from_model）里该是文字的地方写成了
  数组、对象的（列名、表类型、业务日期的列、度量类型，columns 不是数组的）直接丢掉、不抛异常：实测列名写成数组时
  抛 TypeError，整张表记「起草这张表时出错」。
- 补问时请写的列都列出来了、字段留空：表上算拿不准（记留空），用量和提醒里也不再算一次「只给空壳」没写出来；以前
  提醒会因此劝人换模型，真正该做的是补充起草要求。
- 留空的列不算目录里已经有中文名或含义的（不论来源：以前模型起草、按「重新起草不丢内容」留下的，库注释，人工；驳回
  的不算有）；写入时（write_draft）再按并入目录之后的结果筛一遍，起草进行中有人填了的同样不算。以前照界面的建议补了
  起草要求再起草，模型这次没再给的老列又被说成「拿不准、留空了」，提示一直消不掉。
- 一项都没写出来的宽表同样逐段写上列的范围：「这张表 95 列，分 4 段起草，第 1 段（表级的项）、第 2 段（c001 到
  c032，共 32 列）……都没写出来：原因」；只有一段的表照旧只写原因（这一段就是整张表）。
- 提醒照约定写「建议在「设置 → 偏好设置」的「起草模型」换一个模型、缩小起草范围，或者停止这次起草」。

backend/app/data/catalog.py

@@ -1711,7 +1711,8 @@ def _clean_text(value: Any, limit: int = _TEXT_MAX) -> str | None:
 def _from_model(entry: Mapping[str, Any], meta: Mapping[str, Any], *, system: bool) -> dict[str, Any]:
     """模型给的一张表 → 目录项（llm / proposed）。不合规的值整项丢掉，不硬塞：列名对不上表结构的、
     表类型和度量类型不在取值里的、业务主键里有不存在的列的、看起来是计算公式的（_formula_like，公式进口径卡）。
-    导入表格的源，说明已经覆盖的字段也丢掉。"""
+    该是文字的地方写成了数组、对象（列名、表类型、业务日期的列、度量类型……）同样丢掉，不抛异常：实测见过把列名
+    写成数组的，以前在这里抛 TypeError，整张表记出错。导入表格的源，说明已经覆盖的字段也丢掉。"""
     columns = {c["name"]: c for c in meta.get("columns") or []}
     notes: dict[str, Any] = {}
     table_has_system_note = system and bool((meta.get("comment") or "").strip())
@@ -1720,21 +1721,23 @@ def _from_model(entry: Mapping[str, Any], meta: Mapping[str, Any], *, system: bo
             continue
         if (text := _clean_text(entry.get(name))) is not None and not _formula_like(("t", name), text):
             notes[name] = make_item(text, "llm")
-    if entry.get("kind") in TABLE_KINDS:
-        notes["kind"] = make_item(entry["kind"], "llm")
+    kind = entry.get("kind")
+    if isinstance(kind, str) and kind in TABLE_KINDS:
+        notes["kind"] = make_item(kind, "llm")
     keys = entry.get("keys")
     if _str_list_ok(keys) and all(k in columns for k in keys):
         notes["keys"] = make_item(list(dict.fromkeys(keys)), "llm")
     date = entry.get("business_date")
-    if isinstance(date, dict) and date.get("column") in columns:
+    if isinstance(date, dict) and isinstance(date.get("column"), str) and date["column"] in columns:
         value = {"column": date["column"]}
         for k in ("rule", "timezone"):
             if (text := _clean_text(date.get(k))) is not None:
                 value[k] = text
         notes["business_date"] = make_item(value, "llm")
     out_cols: dict[str, dict[str, Any]] = {}
-    for col in entry.get("columns") or []:
-        if not isinstance(col, dict) or col.get("name") not in columns:
+    listed = entry.get("columns")
+    for col in listed if isinstance(listed, list) else []:
+        if not isinstance(col, dict) or not isinstance(col.get("name"), str) or col["name"] not in columns:
             continue
         name = col["name"]
         has_system_note = system and bool((columns[name].get("comment") or "").strip())
@@ -1745,8 +1748,9 @@ def _from_model(entry: Mapping[str, Any], meta: Mapping[str, Any], *, system: bo
             text = _clean_text(col.get(field_name), limit)
             if text is not None and not _formula_like(("c", name, field_name), text):
                 items[field_name] = make_item(text, "llm")
-        if col.get("measure") in COLUMN_MEASURES:
-            items["measure"] = make_item(col["measure"], "llm")
+        measure = col.get("measure")
+        if isinstance(measure, str) and measure in COLUMN_MEASURES:
+            items["measure"] = make_item(measure, "llm")
         if items:
             out_cols[name] = items
     if out_cols:
@@ -1806,7 +1810,8 @@ class TableDraftResult:
     status: str = "ok"
     #: 模型分几段起草（不用模型时为 None）
     parts_total: int | None = None
-    #: 模型拿不准、留空了的列（catalog_draft.ModelOutcome.blank，按表结构的顺序）；表的状态不因此变
+    #: 模型拿不准、留空了的列（catalog_draft.ModelOutcome.blank，按表结构的顺序），只留并入目录之后仍然没有中文名和
+    #: 含义的；表的状态不因此变
     blank: list[str] = field(default_factory=list)
 
 
@@ -1839,6 +1844,8 @@ async def write_draft(session: AsyncSession, source: Any, table: str, modeled: A
     - 并入已有目录按 merge_notes 的规则：人工确认、驳回过的项不动。内容没变不写、不升版本。
     - 模型这次没给的项（留空、没通过校验）不删以前起草的；列已经不在表结构里的，模型各段都写出来时（covered 含 llm）
       照旧删掉。模型只写出一部分的不算完整起草过（covered 不含 llm）。
+    - 模型留空的列（blank）按并入之后的目录再筛一遍：已经有中文名或含义的（以前起草的、库注释、起草进行中有人填的）
+      不算留空。
     - 写入撞上别人刚改过（乐观锁）就重读重并，最多三次。
     """
     result = TableDraftResult(table=table)
@@ -1875,6 +1882,10 @@ async def write_draft(session: AsyncSession, source: Any, table: str, modeled: A
     else:
         result.error = WRITE_CONFLICT_ERROR
         result.added = result.updated = result.removed = 0
+        result.blank = []
+    if result.blank:
+        named = visible_notes(merged).get("columns") or {}
+        result.blank = [c for c in result.blank if not any(f in (named.get(c) or {}) for f in ("label", "meaning"))]
     if result.error or (modeled is not None and modeled.status == "failed"):
         result.status = "failed"
     elif modeled is not None and modeled.status == "partial":

backend/app/data/catalog_draft.py

@@ -30,13 +30,16 @@
   不等于额度用完：只有停在额度上、或者正文为空而有思考才算。
 - 被截断的 JSON：先把已经完整的表、列对象捡回来，只补剩下的。
 - 一段里缺了的表、列只补缺的，补一次。补问之后：模型列出了、字段留空的列是拿不准（补问时才列出来的也一样），不算
-  没写出来；整列没列出来的算没写出来（这张表是 partial）；整张表还一项都没有的，照只给空壳处理（拆小、再问一次）。
+  没写出来（补问的那次调用把请写的都列出来了，用量和提醒里也不算没写出来）；整列没列出来的算没写出来（这张表是
+  partial）；整张表还一项都没有的，照只给空壳处理（拆小、再问一次）。
 - 结构化输出这个接口不支持（400 / 422 提到 response_format、json_schema、tool……，或者模型类不支持）：这次起草
   后面改走纯文本、从正文取 JSON（纯文本那条路另外说清 JSON 的格式）。
 - 服务方嫌额度太大：退回 MODEL_FALLBACK_MAX_TOKENS 重来，这次起草后面都用它（catalog.DraftModel.shrink）。
 - 网络错误、429、5xx：等一会儿重试（最多 MODEL_RETRIES 次，退避）。单次调用的时限是 MODEL_TIMEOUT_S。SDK
   自己不再重试（catalog.resolve_draft_model 建模型时 max_retries=0），实际重试次数和报错写的一致。
 - 别的错误：这一段没写出来，原因照实写。
+- 读回复时出了意料之外的错：这一块记没写出来（「起草这张表时出错」），这次调用照样记进用量、算一次出错。该是文字的
+  地方写成了数组、对象的值在校验（catalog._from_model）里丢掉，不算出错。
 - 封顶：每个原始段（计划里的整表段、表级那段或一个列分段，连同按缩小后的上限分开的各块、拆出来的、补问的、重试的）
   一共最多调用 MODEL_PART_MAX_CALLS 次，到了次数还没写完的记没写出来。
 - 止损：连续 MODEL_GIVE_UP_AFTER 段拆到底仍一项没写出来（模型明显用不了，例如思考总是吃满额度），这次起草后面的
@@ -45,14 +48,18 @@
   超时的不再拆小，超时拆出来、还没问的块也不再问；有一次调用写出来了就恢复。不然一直超时的模型首轮同时起草的几段
   要各自拆满次数，一段 40 列串着问约 7 个时限（420 秒时约 49 分钟）。
 - 提醒：最近 MODEL_NOTICE_WINDOW 次调用里有 MODEL_NOTICE_ON 次以上没写出来（额度用完、超时、空壳、出错），给一句
-  提醒（次数和原因，建议换起草模型、减少表或停止），经进度回调交给任务；之后少于 MODEL_NOTICE_OFF 次时清掉。以前
-  看不出调用在反复失败：一个任务 55 分钟只结束 1 张表，界面上只有段进度和用量。
+  提醒（次数和原因，建议换起草模型、缩小起草范围或停止），经进度回调交给任务；之后少于 MODEL_NOTICE_OFF 次时清掉。
+  以前看不出调用在反复失败：一个任务 55 分钟只结束 1 张表，界面上只有段进度和用量。
 
-**原因。** 没写出来的原因只写进这张表的 model_error：几段因为同一个原因没写出来时说「这几段」，每段写上列的范围
-（「第 5 段（列名 到 列名，共 33 列）」，列名照表结构原样），真拆小重试过的才补「已拆小重试仍没写出」。
+**原因。** 没写出来的原因只写进这张表的 model_error：几段因为同一个原因没写出来时说「这几段」，分段起草的表（只写
+出一部分的、一项都没写出来的都一样）每段写上列的范围（「第 5 段（列名 到 列名，共 33 列）」，列名照表结构原样），
+真拆小重试过的才补「已拆小重试仍没写出」。只有一段的表，这一段就是整张表：只写出一部分的写清是哪几列，一项都没写
+出来的只写原因。
 
 **留空的列。** 请模型写了、模型也列出了，中文名和含义却都没给（或者给的都没通过校验）的列是模型拿不准，不补问，
-记在 ModelOutcome.blank 里（表的状态不因此变）。已确认的、库注释已经说清的列不算（提示词里它们以此为准）。
+记在 ModelOutcome.blank 里（表的状态不因此变）。目录里已经有中文名或含义的列不算（不论来源，驳回的不算有：以前模型
+起草的留着，照「补充起草要求再重新起草」做一遍，模型这次没再给的老列不该又说成拿不准），库注释已经说清的不算
+（提示词里它们以此为准）；写入时（catalog.write_draft）再按并入目录之后的结果筛一遍。
 
 **并发。** 全进程最多 MODEL_CONCURRENCY 个调用同时进行（model_gate：按事件循环各一个信号量，所有起草任务和同步
 起草共用，先来先放）。一次起草里同时只起草 MODEL_CONCURRENCY 段（一段拆小、补缺都做完再起草下一段），同时在等
@@ -121,7 +128,7 @@ REASON_HALTED = "；模型连续 {n} 段都没写出来，已停止拆小重试
 REASON_SLOW = "；拆到每段 {n} 列仍然超时，已停止拆小重试，建议更换起草模型"
 
 # ---- 提醒（最近几次调用多数没写出来），经进度回调交给任务（CatalogDraftJob.notice）
-NOTICE = ("最近 {calls} 次调用有 {failed} 次没写出来（{kinds}）。可以在{place}换一个模型、减少这次起草的表，"
+NOTICE = ("最近 {calls} 次调用有 {failed} 次没写出来（{kinds}）。建议在{place}换一个模型、缩小起草范围，"
           "或者停止这次起草")
 #: 一次调用没写出来的原因种类 → 提醒里的说法（按这个顺序列）
 FAILURE_LABEL = {"exhausted": "额度用完", "timeout": "超时", "hollow": "只给空壳", "error": "出错"}
@@ -980,14 +987,22 @@ class _Run:
                 return
             self._fail(pieces, _Why(reply.error, split=halved))
             return
-        gains = self._take(pieces, reply.entries)
+        try:
+            gains = self._take(pieces, reply.entries)
+        except Exception:
+            # 读回复时出了意料之外的错（_guarded 记这一块没写出来）：这次调用照样记进用量，算一次出错
+            await self._called(reply.usage or (0, 0), "error")
+            raise
         anything = any(g.items for g in gains)
-        await self._report(reply, wrote=anything)
+        complete = not (reply.exhausted or reply.truncated)
+        rest = self._remainder(pieces, gains, truncated=False, followup=followup) if complete else []
+        # 补问时请写的都列出来了、字段留空：是模型拿不准（记留空），这次调用不算没写出来（不进提醒的窗口）
+        await self._report(reply, wrote=anything or (followup and complete and not rest))
         if last and anything and self.mode != "text":
             # 结构化输出只给空壳、纯文本写出来了：这个模型的结构化输出靠不住，这次起草后面都走纯文本
             self.mode = "text"
             logger.info("起草数据目录：结构化输出只给了空壳，纯文本写出来了，后面改走纯文本")
-        if reply.exhausted or reply.truncated:
+        if not complete:
             rest = self._remainder(pieces, gains, truncated=True, followup=followup)
             if not rest:
                 return
@@ -1007,7 +1022,6 @@ class _Run:
                 return
             await self._split(pieces, depth, "hollow", followup, budget, halved)
             return
-        rest = self._remainder(pieces, gains, truncated=False, followup=followup)
         if not rest:
             return
         if followup:
@@ -1333,6 +1347,11 @@ class _Run:
         if not items:
             failures = [(why, part_of.get(key, 1)) for key, why in st.failed.items()]
             reason = _join(failures) if failures else (C.MODEL_EMPTY_REASON if st.named else REASON_MISSING)
+            if failures and len(st.parts) > 1:
+                # 宽表一项都没写出来：同样逐段写上列的范围
+                said = self._parts_said(st, failed_cols, failed_table)
+                every = "都" if len(said) == len(st.parts) else ""
+                reason = f"这张表 {len(st.columns)} 列，分 {len(st.parts)} 段起草，{'、'.join(said)}{every}没写出来：{reason}"
             return ModelOutcome("failed", {}, 0, reason, total, st.parts_done)
         notes = copy.deepcopy(st.notes)
         blank = self._blank(st, set(failed_cols))
@@ -1342,18 +1361,21 @@ class _Run:
         return ModelOutcome("ok", notes, items, None, total, st.parts_done, blank)
 
     def _blank(self, st: _Table, failed: set[str]) -> list[str]:
-        """模型拿不准、留空了的列：请它写了（不在没写出来的里面），中文名和含义却都没给（或者给的都没通过校验）。已确认
-        的、库注释已经说清的不算——提示词里它们以此为准，模型不写也不缺：已确认（有确证）的中文名或含义、库注释（按
-        catalog.comment_role 记进目录的那种；导入表格的源是系统说明）。"""
+        """模型拿不准、留空了的列：请它写了（不在没写出来的里面），中文名和含义却都没给（或者给的都没通过校验）。
+
+        目录里已经有中文名或含义的不算（不论来源：以前模型起草的——重新起草时模型这次没给的留着、不删；人工填写、确认的；
+        库注释；驳回的不算有）：照「补充起草要求再重新起草」做一遍，模型这次没再给的老列不该又说成拿不准。库注释已经说清
+        的（catalog.comment_role 记进目录的那种；导入表格的源是系统说明）同样不算：提示词里它们以此为准，模型不写也不缺。
+        写入时（catalog.write_draft）再按并入目录之后的结果筛一遍。"""
         drafted = st.notes.get("columns") or {}
-        confirmed = C._confirmed_only(st.existing).get("columns") or {}
+        known = C.visible_notes(st.existing).get("columns") or {}
         comments = {str(c.get("name")): (c.get("comment") or "").strip()
                     for c in st.meta.get("columns") or [] if isinstance(c, dict)}
         out: list[str] = []
         for col in st.columns:
             if col in failed:
                 continue
-            if any(f in (drafted.get(col) or {}) or f in (confirmed.get(col) or {}) for f in ("label", "meaning")):
+            if any(f in (drafted.get(col) or {}) or f in (known.get(col) or {}) for f in ("label", "meaning")):
                 continue
             text = comments.get(col)
             if text and (self.system or C.comment_role(text, col) is not None):
@@ -1362,18 +1384,25 @@ class _Run:
         return out
 
     @staticmethod
-    def _partial_text(st: _Table, failed_cols: list[str], failed_table: bool) -> str:
+    def _parts_said(st: _Table, failed_cols: list[str], failed_table: bool) -> list[str]:
+        """分段起草的表没写出来的是哪几段，每段写上列的范围（列名照表结构原样）：「第 1 段（表级的项）」「第 3 段（列名 到
+        列名，共 32 列）」。"""
+        part_of = st.part_of()
+        bad: dict[int, list[str]] = {}
+        for col in failed_cols:
+            bad.setdefault(part_of.get(col, 1), []).append(col)
+        said = ["第 1 段（表级的项）"] if failed_table else []
+        return said + [f"第 {k} 段（{_columns_text(st.columns, cols, count=True)}）" for k, cols in sorted(bad.items())]
+
+    @classmethod
+    def _partial_text(cls, st: _Table, failed_cols: list[str], failed_table: bool) -> str:
         """只写出一部分时说清哪部分没写出来（第几段、哪几列）、为什么、怎么办。"""
         part_of = st.part_of()
         keys = ([_TABLE] if failed_table else []) + failed_cols
         reason = _join((st.failed[k], part_of.get(k, 1)) for k in keys)
         n = len(st.columns)
         if len(st.parts) > 1:
-            bad: dict[int, list[str]] = {}
-            for col in failed_cols:
-                bad.setdefault(part_of.get(col, 1), []).append(col)
-            said = ["第 1 段（表级的项）"] if failed_table else []
-            said += [f"第 {k} 段（{_columns_text(st.columns, cols, count=True)}）" for k, cols in sorted(bad.items())]
+            said = cls._parts_said(st, failed_cols, failed_table)
             head = f"这张表 {n} 列，分 {len(st.parts)} 段起草，{'、'.join(said)}没写出来"
         elif failed_cols and failed_table:
             head = (f"这张表 {n} 列，有表级的项和 {len(failed_cols)} 列（{_columns_text(st.columns, failed_cols, count=False)}）"

backend/tests/test_catalog_draft_engine.py

@@ -462,11 +462,13 @@ async def test_columns_listed_blank_in_the_followup_are_left_blank_not_failed():
             entry["columns"] = [{"name": c["name"]} for c in entry["columns"]]
         return reply
 
-    model = Scripted(handle)
-    got = await D.run_model_draft(model, _source(tables), ["t8"])
+    model, reports = Scripted(handle), Reports()
+    got = await D.run_model_draft(model, _source(tables), ["t8"], hooks=reports.hooks())
     assert model.calls[1].asked == {"t8": ("columns", ["c003", "c007"])}
     o = got["t8"]
     assert o.status == "ok" and o.error is None and o.blank == ["c003", "c007"]
+    # 补问的那次把请写的列都列出来了（字段留空是拿不准）：用量和提醒里也不算一次「只给空壳」
+    assert reports.failures == [None, None]
 
 
 async def test_a_followup_that_comes_back_as_an_empty_shell_leaves_the_table_partial():
@@ -832,7 +834,8 @@ async def test_the_same_reason_is_said_once_even_when_some_parts_were_split(monk
     got = await D.run_model_draft(model, _source({"wide45": _table(45)}), ["wide45"])
     o = got["wide45"]
     assert o.status == "failed"
-    assert o.error == "模型只给了空壳，没有写出可用的内容；已拆小重试仍没写出"
+    assert o.error == ("这张表 45 列，分 3 段起草，第 1 段（表级的项）、第 2 段（c001 到 c023，共 23 列）、"
+                       "第 3 段（c024 到 c045，共 22 列）都没写出来：模型只给了空壳，没有写出可用的内容；已拆小重试仍没写出")
 
 
 def test_reasons_are_joined_by_their_base():
@@ -894,7 +897,8 @@ async def test_the_column_parts_of_a_wide_table_stop_splitting_too(monkeypatch):
     model = Scripted(_hollow)
     got = await D.run_model_draft(model, _source({"wide95": _table(95)}), ["wide95"])
     assert len(model.calls) == 2 + 8 + 2 + 2
-    assert got["wide95"].error == (D.REASON_HOLLOW + "；每段最多调用 8 次模型，这一段到了次数仍没写完，已停止重试"
+    assert got["wide95"].error == (WIDE95_PARTS + "都没写出来：" + D.REASON_HOLLOW
+                                   + "；每段最多调用 8 次模型，这一段到了次数仍没写完，已停止重试"
                                    + D.REASON_SPLIT + "；模型连续 2 段都没写出来，已停止拆小重试，建议更换起草模型")
 
 
@@ -938,6 +942,11 @@ async def test_a_big_draft_does_not_hold_the_model_ahead_of_a_later_one(monkeypa
     assert order == ["a1", "b1", "a2", "a3", "a4"]
 
 
+#: 95 列的宽表分 4 段起草，逐段写上列的范围
+WIDE95_PARTS = ("这张表 95 列，分 4 段起草，第 1 段（表级的项）、第 2 段（c001 到 c032，共 32 列）、"
+                "第 3 段（c033 到 c064，共 32 列）、第 4 段（c065 到 c095，共 31 列）")
+
+
 def _order(model: Scripted) -> list[tuple[str, str]]:
     return [(name, part) for call in model.calls for name, (part, _) in call.asked.items()]
 
@@ -1008,8 +1017,8 @@ async def test_the_notice_comes_when_four_of_six_calls_fail_and_goes_when_fewer_
     assert reports.failures == ["error", "error", "exhausted", "hollow", "hollow"] + [None] * 5
     notices = [r.notice for r in reports.items]
     assert notices[:3] == [None] * 3
-    assert notices[3] == ("最近 4 次调用有 4 次没写出来（额度用完 1 次、只给空壳 1 次、出错 2 次）。可以在「设置 → 偏好设置」"
-                          "的「起草模型」换一个模型、减少这次起草的表，或者停止这次起草")
+    assert notices[3] == ("最近 4 次调用有 4 次没写出来（额度用完 1 次、只给空壳 1 次、出错 2 次）。建议在「设置 → 偏好设置」"
+                          "的「起草模型」换一个模型、缩小起草范围，或者停止这次起草")
     assert notices[4].startswith("最近 5 次调用有 5 次没写出来")
     assert notices[7].startswith("最近 6 次调用有 3 次没写出来"), "还有 2 次以上没写出来：留着，次数照实写"
     assert notices[8].startswith("最近 6 次调用有 2 次没写出来")
@@ -1018,7 +1027,8 @@ async def test_the_notice_comes_when_four_of_six_calls_fail_and_goes_when_fewer_
 
 
 # --------------------------------------------------------------------------
-# 复核后的修正：每个原始段一共 8 次、一直超时的模型几轮就停、止损后纯文本不再原样再问
+# 复核后的修正：每个原始段一共 8 次、一直超时的模型几轮就停、读回复出错也记用量、目录里已有的不算留空、补问列出了不算
+# 没写出来、一项都没写出来的宽表逐段写范围、止损后纯文本不再原样再问
 # --------------------------------------------------------------------------
 
 
@@ -1115,6 +1125,77 @@ async def test_timeouts_that_go_away_at_ten_columns_are_still_split_down_and_wri
     assert len(model.calls) == 7 and max(slow.rounds) == 4
 
 
+async def test_odd_values_in_a_reply_are_dropped_not_a_crash():
+    """模型把列名、表类型、业务主键、业务日期的列、度量类型写成了数组或对象：这几项丢掉，别的照收；不当成「起草这张表
+    时出错」。以前列名是数组时校验里抛 TypeError，整张表记出错，这次调用也没有记进用量。"""
+    tables = {"t4": _table(4)}
+
+    def handle(call: Call) -> Reply:
+        return Reply(obj={"tables": [{
+            "table": "t4", "label": "某表", "kind": ["fact"], "keys": [["c001"]],
+            "business_date": {"column": ["c002"], "rule": "按下单时间"},
+            "columns": [{"name": ["c001"], "label": "数组列名"}, {"name": {"x": 1}, "label": "对象列名"},
+                        {"name": "c002", "label": "第二列", "measure": ["flow"]},
+                        {"name": "c003", "label": "第三列", "measure": {"x": "flow"}},
+                        {"name": "c004", "label": "第四列"}]}]})
+
+    model, reports = Scripted(handle), Reports()
+    got = await D.run_model_draft(model, _source(tables), ["t4"], hooks=reports.hooks())
+    o = got["t4"]
+    assert len(reports.items) == len(model.calls) == 2, "c001 没有按列名列出来：补问一次"
+    assert o.status == "partial" and "c001" in o.error and "出错" not in o.error
+    assert set(o.notes) == {"label", "columns"} and set(o.notes["columns"]) == {"c002", "c003", "c004"}
+    assert all(set(items) == {"label"} for items in o.notes["columns"].values())
+
+
+async def test_a_call_whose_reply_cannot_be_read_still_counts(monkeypatch):
+    """读回复时出了意料之外的错：这张表记出错，这次调用照样记进用量（算一次出错），调用次数和用量对得上。"""
+    tables = {"t4": _table(4)}
+
+    def broken(entry, meta, *, system):
+        raise RuntimeError("校验时出了意外")
+
+    monkeypatch.setattr(catalog, "_from_model", broken)
+    model, reports = Scripted(answering(tables, usage=(100, 50))), Reports()
+    got = await D.run_model_draft(model, _source(tables), ["t4"], hooks=reports.hooks())
+    assert len(model.calls) == 1 and reports.failures == ["error"]
+    assert [(r.input_tokens, r.output_tokens) for r in reports.items] == [(100, 50)]
+    assert got["t4"].status == "failed" and got["t4"].error == "起草这张表时出错：校验时出了意外"
+
+
+async def test_columns_the_catalog_already_names_are_not_counted_blank():
+    """模型这次留空、目录里已经有中文名或含义的列（不论来源：以前模型起草的、库注释、人工；驳回的不算）不算留空：照界面
+    的建议补了起草要求再起草，以前起草好的留着（重新起草不丢内容），这一列不该再说成「模型拿不准、留空了」。"""
+    tables = {"t6": _table(6)}
+    item = catalog.make_item
+    existing = {"t6": {"columns": {
+        "c002": {"label": item("以前起草的名字", "llm")},
+        "c003": {"meaning": item("库注释说清的含义", "comment")},
+        "c004": {"label": item("驳回的名字", "llm", "rejected")},
+        "c005": {"unit": item("元", "llm")},
+    }}}
+
+    def handle(call: Call) -> Reply:
+        cols = [{"name": "c001", "label": "第一列"}] + [{"name": f"c00{i}"} for i in range(2, 7)]
+        return Reply(obj={"tables": [{"table": "t6", "label": "某表", "columns": cols}]})
+
+    got = await D.run_model_draft(Scripted(handle), _source(tables), ["t6"], existing=existing)
+    assert got["t6"].status == "ok" and got["t6"].blank == ["c004", "c005", "c006"]
+
+
+async def test_a_wide_table_that_wrote_nothing_names_every_part():
+    """一项都没写出来的宽表同样逐段写上列的范围（以前只写原因和「这一段」「这几段」，看不出是哪几段）。"""
+    def handle(call: Call) -> Reply:
+        raise RuntimeError("网关拒绝")
+
+    got = await D.run_model_draft(Scripted(handle), _source({"wide95": _table(95)}), ["wide95"])
+    o = got["wide95"]
+    assert o.status == "failed" and o.error == WIDE95_PARTS + "都没写出来：调用模型出错：网关拒绝"
+
+    narrow = await D.run_model_draft(Scripted(handle), _source({"t8": _table(8)}), ["t8"])
+    assert narrow["t8"].error == "调用模型出错：网关拒绝", "只有一段的表：这一段就是整张表，照旧只写原因"
+
+
 async def test_a_halted_part_that_could_be_split_is_not_asked_again_in_text_mode(monkeypatch):
     """止损以后本来能拆小的段只给空壳：已经走纯文本的不再原样再问（约定只对拆不开的段再问一次）；结构化输出的照旧换
     纯文本问一次，看看是不是结构化输出的问题（见 test_the_column_parts_of_a_wide_table_stop_splitting_too）。"""

backend/tests/test_catalog_draft_jobs.py

@@ -287,9 +287,9 @@ async def test_a_redraft_keeps_what_the_model_left_blank_this_time(client, monke
     给了合格新值的才替换。以前整库重跑一次删掉几百项，表仍记「已完成」。"""
     sid = await _source(_tickets())
     full = {"label": "门票", "kind": "fact",
-            "columns": [{"name": f"c{i:02d}", "label": f"第{i}列", "meaning": f"第{i}列的含义"} for i in range(1, 7)]}
+            "columns": [{"name": f"c{i:02d}", "label": f"第{i}列", "meaning": f"第{i}列的含义"} for i in range(1, 6)]}
     first = await _draft(client, sid, monkeypatch, FakeModel({"tickets": full}), ["tickets"])
-    assert first["tasks"][0]["status"] == "ok" and first["blank_columns"] == 0
+    assert first["tasks"][0]["status"] == "ok" and first["blank_columns"] == 1, "c06 第一次就留空了"
     before = (await client.get(f"{_base(sid)}/tickets")).json()["notes"]
 
     # 这一次：表的中文名换了；c03 的中文名像公式（校验丢掉）、含义换了；c04 只给了中文名；其余列列出了但都留空
@@ -303,10 +303,11 @@ async def test_a_redraft_keeps_what_the_model_left_blank_this_time(client, monke
     assert after["label"]["value"] == "门票明细" and after["kind"] == before["kind"]
     assert after["columns"]["c03"]["label"] == before["columns"]["c03"]["label"]
     assert after["columns"]["c03"]["meaning"]["value"] == "新的含义"
-    for col in ("c01", "c02", "c05", "c06"):
+    for col in ("c01", "c02", "c05"):
         assert after["columns"][col] == before["columns"][col], col
-    # 模型这次留空的列照实记下（表的状态不因此变）
-    assert task["blank_columns"] == 4 and task["blank_sample"] == ["c01", "c02", "c05", "c06"]
+    # 留空的列只记目录里还是没有中文名、含义的（c06）：c01、c02、c05 这次留空，以前起草好的还在，不算留空——不然照
+    # 提示补了起草要求再起草，提示指着的却是目录里有中文名的列，一直消不掉
+    assert task["blank_columns"] == 1 and task["blank_sample"] == ["c06"]
 
 
 async def test_model_items_of_a_column_gone_from_the_schema_are_removed(client, monkeypatch):
@@ -351,8 +352,8 @@ async def test_the_notice_is_kept_on_the_job(client, monkeypatch):
     replies = {name: {"label": f"表{name}"} for name in tables}
     job = await _draft(client, sid, monkeypatch, FakeModel(replies, fail={"t1", "t2", "t3", "t4"}), list(tables))
     assert job["usage"]["calls"] == 5 and job["usage"]["failed_calls"] == 4
-    assert job["notice"] == ("最近 5 次调用有 4 次没写出来（出错 4 次）。可以在「设置 → 偏好设置」的「起草模型」换一个模型、"
-                             "减少这次起草的表，或者停止这次起草")
+    assert job["notice"] == ("最近 5 次调用有 4 次没写出来（出错 4 次）。建议在「设置 → 偏好设置」的「起草模型」换一个模型、"
+                             "缩小起草范围，或者停止这次起草")
     plain = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["t1"]})).json()["job"]
     await finished(plain["id"])
     plain = await get_job(client, sid, plain["id"])
@@ -375,6 +376,29 @@ async def test_blank_columns_reach_the_task_the_job_and_the_list(client, monkeyp
     assert rows["tickets"]["last_draft"]["blank_columns"] == 7 and rows["gates"]["last_draft"]["blank_columns"] == 1
 
 
+async def test_blank_columns_are_counted_against_the_catalog_as_written(client, monkeypatch):
+    """留空的列按并入目录之后的结果算：起草进行中有人给某一列填了中文名，模型这次留空，这一列不算留空。"""
+    sid = await _source(_tickets(4))
+    model = Held({"tickets": {"label": "门票", "columns": [{"name": "c01", "label": "编号"}]}}, hold={"tickets"})
+    use(monkeypatch, model)
+    r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["tickets"], "use_model": True})
+    job_id = r.json()["job"]["id"]
+
+    async def asking() -> bool:
+        return (await get_job(client, sid, job_id))["tasks"][0]["status"] == "running"
+
+    await until(asking)
+    saved = await client.put(f"{_base(sid)}/tickets", json={"notes": {"columns": {"c02": {"label": {"value": "闸机号"}}}},
+                                                           "if_version": 0})
+    assert saved.status_code == 200, saved.text
+    model.release.set()
+    await finished(job_id)
+    task = (await get_job(client, sid, job_id))["tasks"][0]
+    assert task["status"] == "ok" and task["blank_columns"] == 2 and task["blank_sample"] == ["c03", "c04"]
+    after = (await client.get(f"{_base(sid)}/tickets")).json()["notes"]["columns"]
+    assert after["c02"]["label"]["value"] == "闸机号" and after["c01"]["label"]["value"] == "编号"
+
+
 async def test_the_sync_draft_is_409_while_a_job_is_queued_or_running(client, monkeypatch):
     """旧的同步起草不和后台任务同时写同一个源的目录、抢模型名额：有排队或进行中的任务时 409（和建任务同一句话）。"""
     sid = await _source()