认证队列的确定性重复按规范形分组：条件换序、AND 两边换位的进同一组

以前按 normalize_sql 的规范文本分组，WHERE 的条件照原文的先后，同一个形状换了条件顺序就拆成两组，凑不够门槛进不了队。

- sql_semantics.sql_shape：在判等价同一套不认列的规范形式（_light）上把字面量抹成只看类型（字还是数）的占位，再把交换律
  成立的地方按抹掉之后的写法重排——AND / OR 的各项、IN 列表、= / <> / + / * 的两边、比大小的两边（换边时大于小于跟着
  换）；抹掉之后一样的几项按原值排，同一组里字面量的位置对得上。不另写解析器。认不出的（解析不了、sqlglot 会读错的写法、
  字面量对不回 normalize_sql 的）为 None，照旧按规范文本分组（只会分得细）。positions 把规范形里的字面量对回 normalize_sql
  的位置，固化成模板时用。
- ledger/certify：Shape 加 key / canonical / positions；队列按（数据源、规范形）分组，跨轮变了的位置按规范形对齐，再换回
  代表那条的位置；固化时同组的别的运行也按规范形认、按同样的对齐认参数。这一项的 id 按规范形算；以前按规范文本记的
  「不是口径」照样认（组里任何一条的规范文本对得上就算）。

测试：tests/test_ledger_certify.py 新增四条——三种换序写法同一个形状、字面量位置对得上；七种真不同（换列、OR、少条件、
换运算符、数对字、IN 列表、换聚合）不是同一个形状；真跑三轮换序的问数据进同一组、多筛一个地区的不进，固化出两个日期
参数；按旧 id 记的「不是口径」照样生效。把分组改回按文本再跑，换序那条失败。

backend/app/data/sql_semantics.py

@@ -9161,8 +9161,168 @@ def output_names(sql: str, dialect: str | None = None, *, template: bool = False
     return [_output_name(i, read) for i in root.expressions]
 
 
+# ==========================================================================
+# 同一个形状（认证队列的「确定性重复」：条件的先后不算）
+# ==========================================================================
+#
+# normalize_sql 的规范文本把字面量换成「?」，但条件照原文的先后：WHERE a = 1 AND b = 2 和 WHERE b = 2 AND a = 1 文本不同，
+# 认证队列就拆成两组。这里在不认列的那一级规范形式（_light，和判等价同一套只保持语义的改写）上，把字面量抹成只看类型的
+# 占位，再把交换律成立的地方按抹掉之后的写法重排一遍：AND / OR 的各项、IN 列表、= / <> / + / * 的两边、比大小的两边（换边
+# 时大于小于跟着换）。规范形式里本来的排序看字面量的值（同一个形状换了参数，条件的先后可能就变了），所以要重排。重排时
+# 抹掉之后一样的几项按原值排：同一组里字面量的位置对得上。认不出的（解析不了、sqlglot 会读错的写法、规范化不了的结构）
+# 为 None，调用方照旧按 normalize_sql 的文本分组：不同就是不同，只会分得细，不会把不同的并成一组。
+
+#: 抹掉字面量之前给每个字面量记的编号（对回 normalize_sql 的字面量用）
+_LIT_TAG = "sql_semantics_literal"
+#: 对 normalize_sql 的先后时，数字字面量换成的标记：这个整数加编号（同 _NUMBER_SENTINEL，真实 SQL 里几乎不会写）
+_LIT_NUMBER = 9_173_402_581_000
+_LIT_STRING = "__agl_l{}__"
+_LIT_STRING_RE = re.compile(r"^'__agl_l(\d+)__'$")
+#: 交换律成立、两边可以对调的运算
+_SYMMETRIC = ("EQ", "NEQ", "Add", "Mul", "NullSafeEQ")
+
+
+@dataclass(frozen=True)
+class SqlShape:
+    """sql_shape 的结果。key 一样就是同一个形状（只差字面量的值、条件的先后、交换律成立处两边的先后）；literals 是按规范
+    形式里的先后排的字面量（SQL 写法，字符串带引号），同一个形状的几条按位置对得上；positions[i] 是第 i 个在
+    normalize_sql(sql).literals 里的位置（规范化时新算出来的字面量对不上，为 None）。"""
+
+    key: str
+    literals: tuple[str, ...]
+    positions: tuple[int | None, ...]
+
+
+def _normalized_order(root: Any, read: str | None) -> list[int] | None:
+    """normalize_sql 的字面量（按它重新生成的文本里的先后）各是哪一个字面量节点（_LIT_TAG 的编号）：把字面量换成带编号的
+    标记，同样地重新生成、切词，按先后读出编号。对不上（个数不同、有认不出的）为 None。"""
+    from sqlglot import exp
+    from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
+
+    marked = normalize_identifiers(root.copy(), dialect=read)
+    for lit in marked.find_all(exp.Literal):
+        tag = lit.meta.get(_LIT_TAG)
+        if tag is None:
+            return None
+        lit.set("this", _LIT_STRING.format(tag) if lit.is_string else str(_LIT_NUMBER + tag))
+    text = marked.sql(dialect=read, comments=False)
+    out: list[int] = []
+    for token in _tokens(text, read):
+        if not _is_literal(token):
+            continue
+        raw = text[token.start:token.end + 1]
+        found = _LIT_STRING_RE.match(raw)
+        if found:
+            out.append(int(found.group(1)))
+            continue
+        try:
+            out.append(int(raw) - _LIT_NUMBER)
+        except ValueError:
+            return None
+    return out
+
+
+def _shape_print(node: Any, read: str | None) -> tuple[str, list[Any]]:
+    """规范形式的确定写法（同 _erased：字面量只写类型、不写值，按写出的先后收进列表），交换律成立的地方按抹掉之后的写法
+    重排（抹掉之后一样的几项按原值排）。"""
+    from sqlglot import exp
+
+    def value_key(lits: list[Any]) -> tuple[str, ...]:
+        return tuple(lit.sql(dialect=read) for lit in lits)
+
+    def ordered(parts: list[tuple[str, list[Any]]]) -> list[tuple[str, list[Any]]]:
+        return sorted(parts, key=lambda p: (p[0], value_key(p[1])))
+
+    if isinstance(node, exp.Paren):
+        return _shape_print(node.this, read)
+    if isinstance(node, exp.Literal):
+        return f"Literal({'s' if node.is_string else 'n'})", [node]
+    if isinstance(node, (exp.And, exp.Or)):
+        parts = ordered([_shape_print(p, read) for p in _flat(node, type(node))])
+        return (f"{type(node).__name__}[{','.join(p[0] for p in parts)}]",
+                [lit for _, lits in parts for lit in lits])
+    if isinstance(node, exp.Expr) and type(node).__name__ in _FLIP and not _quantified(node.this) \
+            and not _quantified(node.expression):
+        left, right = _shape_print(node.this, read), _shape_print(node.expression, read)
+        name = type(node).__name__
+        if (_fixed(node.this) and not _fixed(node.expression)) or (
+                _fixed(node.this) == _fixed(node.expression) and (left[0], value_key(left[1])) > (right[0], value_key(right[1]))):
+            left, right, name = right, left, _FLIP[name]
+        return f"{name}({left[0]},{right[0]})", left[1] + right[1]
+    if isinstance(node, exp.Expr) and type(node).__name__ in _SYMMETRIC:
+        parts = ordered([_shape_print(node.this, read), _shape_print(node.expression, read)])
+        return f"{type(node).__name__}({','.join(p[0] for p in parts)})", parts[0][1] + parts[1][1]
+    if isinstance(node, exp.In) and node.args.get("expressions"):
+        head = _shape_print(node.this, read)
+        parts = ordered([_shape_print(e, read) for e in node.expressions])
+        rest = [f"{k}={_shape_print(v, read)[0]}" for k, v in sorted(node.args.items())
+                if k not in ("this", "expressions") and v is not None and v is not False and v != []]
+        return (f"In({head[0]};[{','.join(p[0] for p in parts)}];{','.join(rest)})",
+                head[1] + [lit for _, lits in parts for lit in lits])
+    if isinstance(node, exp.Expr):
+        pieces: list[str] = []
+        lits: list[Any] = []
+        for k, v in sorted(node.args.items()):
+            if v is None or v is False or v == []:
+                continue
+            text, found = _shape_print(v, read)
+            pieces.append(f"{k}={text}")
+            lits += found
+        mark = node.meta.get("sql_semantics_join") if isinstance(node, exp.Join) else None
+        if mark:
+            pieces.append(f"@join={mark}")
+        surface = node.meta.get(_SURFACE)
+        if surface:
+            pieces.append(f"@surface={surface}")
+        return f"{type(node).__name__}({','.join(pieces)})", lits
+    if isinstance(node, list):
+        parts = [_shape_print(v, read) for v in node]
+        return "[" + ",".join(p[0] for p in parts) + "]", [lit for _, found in parts for lit in found]
+    return repr(node), []
+
+
+def sql_shape(sql: str, dialect: str | None = None) -> SqlShape | None:
+    """一条查询的形状（见本节开头）：条件的先后、交换律成立处两边的先后、字面量的值都不算；别的不同一律算不同。认不出的为
+    None（调用方按 normalize_sql 的文本分组）。从不抛异常。"""
+    from sqlglot import exp
+
+    read = _read(dialect)
+    text = (_mysql_dashes(sql) if read == "mysql" and isinstance(sql, str) else sql or "").strip()
+    root, _ = _parse(text, read)
+    if root is None:
+        return None
+    try:
+        if _lost_in_parsing(text, read, root) or _respelled(text, read, root):
+            return None
+        for i, lit in enumerate(root.find_all(exp.Literal)):
+            lit.meta[_LIT_TAG] = i
+        order = _normalized_order(root, read)
+        if order is None or len(order) != len(normalize_sql(sql, dialect).literals):
+            return None
+        marked = _mark_surfaces(root, text, read)
+        if read == "sqlite":
+            _mark_comma_joins(root, text, read)
+        if not marked:
+            return None
+        try:
+            tree = _light(root, read, _Schema()).tree
+        except _Unsupported:
+            # 规范化不了的结构（递归 CTE、不确定的函数……）：只统一名字，条件的先后照原文
+            tree = root.copy()
+            _normalize_names(tree, read, _Schema())
+        printed, lits = _shape_print(tree, read)
+    except Exception:  # noqa: BLE001 - 认形状本身的缺陷：当作认不出，调用方按文本分组
+        logger.warning("认 SQL 的形状失败", exc_info=True)
+        return None
+    where = {tag: i for i, tag in enumerate(order)}
+    return SqlShape(hashlib.sha256(printed.encode()).hexdigest(),
+                    tuple(lit.sql(dialect=read) for lit in lits),
+                    tuple(where.get(lit.meta.get(_LIT_TAG)) if lit.meta.get(_LIT_TAG) is not None else None
+                          for lit in lits))
+
+
 __all__ = ["BINDING_LEVELS", "MAX_SQL_CHARS", "PLACEHOLDER", "Aggregate", "ColumnRef", "Computed", "FacetDiff", "Facets",
-           "JoinFacet", "LimitFacet", "NormalizedSql", "Part", "Projection", "SqlDiff", "TemplateBinding",
+           "JoinFacet", "LimitFacet", "NormalizedSql", "Part", "Projection", "SqlDiff", "SqlShape", "TemplateBinding",
            "TemplateMatch", "TemplateSlot", "TimePredicate", "bind_template", "catalog_labels", "diff_facets",
            "explain_sql_diff", "facets", "match_template", "normalize_sql", "output_names", "project_select",
-           "project_template", "render_bound"]
+           "project_template", "render_bound", "sql_shape"]

backend/app/ledger/certify.py

@@ -5,8 +5,11 @@
 
 **认证队列**（GET /api/ledger/certify-queue，搭建者及以上）三种信号：
 - 确定性重复（shape）：问数据的探索运行里执行过、被报告引用（比对工件里标签记着的查询）、又没对上任何口径（none / related）
-  的查询，按（数据源、normalize_sql 的规范文本）分组；出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的进队。
-  跨运行变化的字面量位置是参数候选，不变的是口径常量。展示的是规范文本（字面量都换成了「?」，不带取值）；
+  的查询，按（数据源、规范形）分组：sql_semantics.sql_shape 在判等价同一套规范形式上抹掉字面量、把 AND / OR 的各项、IN 列表、
+  交换律成立的比较两边重排——条件的先后不同、AND 两边换位的进同一组，条件真不同的不进；认不出规范形的按 normalize_sql 的
+  规范文本分组（只会分得细）。出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的进队。跨运行变化的字面量
+  位置（按规范形里的先后对齐）是参数候选，不变的是口径常量。展示的是代表那条的规范文本（字面量都换成了「?」，不带取值）。
+  这一项的 id 按（数据源、规范形）算；以前按规范文本记的「不是口径」照样认（组里任何一条的规范文本对得上就算）；
 - 问题聚类（cluster）：app/chat/questions 全站唯一的聚类（只取问题原文、不含 history、排除系统派生的运行），≥ REPEAT_RUNS
   个问题的类；
 - 被反复用的答案（reused）：导出过 ≥ 2 次（report.exported 审计，第 2 波）、做成过应用（蓝图的来源是这次运行）、在画布里
@@ -90,22 +93,38 @@ def item_id(kind: str, key: str) -> str:
 
 @dataclass(frozen=True)
 class Shape:
-    """一条查询的形状：数据源、规范文本（字面量换成「?」）、字面量（SQL 写法，字符串带引号）、规范化失败的原因。"""
+    """一条查询的形状：数据源、规范文本（字面量换成「?」）、字面量（SQL 写法，字符串带引号）、规范化失败的原因。
+
+    key 是分组用的规范形（sql_semantics.sql_shape，「shape:」开头；认不出时退回规范文本，「text:」开头）；canonical 是按
+    规范形里的先后排的字面量（同一组的几条按位置对得上，认哪些位置跨轮变了用它），positions[i] 是它在 literals 里的位置
+    （规范化时新算出来的字面量为 None：没法写成模板参数）。"""
 
     source: str
     text: str
     literals: tuple[str, ...]
     problem: str | None = None
+    key: str = ""
+    canonical: tuple[str, ...] = ()
+    positions: tuple[int | None, ...] = ()
+
+    def params(self, canonical_positions: Sequence[int]) -> list[int]:
+        """规范形里的位置 → literals 里的位置（template_of 要的）；对不上的不要。"""
+        return sorted(p for i in canonical_positions if i < len(self.positions) and (p := self.positions[i]) is not None)
 
 
 def shape_of(snap: Any, kinds: dict[str, str | None]) -> Shape | None:
-    from app.data.sql_semantics import normalize_sql
+    from app.data.sql_semantics import normalize_sql, sql_shape
 
     if not isinstance(snap, dict) or not isinstance(snap.get("sql"), str) or not snap["sql"].strip():
         return None
     source = str(snap.get("source") or "")
     norm = normalize_sql(snap["sql"], kinds.get(source))
-    return Shape(source, norm.text, norm.literals, norm.parse_error)
+    found = sql_shape(snap["sql"], kinds.get(source)) if not norm.parse_error else None
+    if found is None:
+        return Shape(source, norm.text, norm.literals, norm.parse_error, key=f"text:{norm.text}",
+                     canonical=norm.literals, positions=tuple(range(len(norm.literals))))
+    return Shape(source, norm.text, norm.literals, None, key=f"shape:{found.key}", canonical=found.literals,
+                 positions=found.positions)
 
 
 def varying(literals: Sequence[Sequence[str]]) -> list[int]:
@@ -174,11 +193,13 @@ def template_of(shape: Shape, params: Sequence[int], dialect: str | None, *,
 class _Member:
     run_id: str
     artifact: str
+    #: 按规范形里的先后排的字面量（Shape.canonical）
     literals: tuple[str, ...]
     conversation_id: str | None
     turn_id: str | None
     question: str
     related: list[dict[str, Any]] = field(default_factory=list)
+    shape: Shape | None = None
 
 
 def referenced_queries(content: dict[str, Any]) -> list[str]:
@@ -297,31 +318,38 @@ async def queue(session: Any, *, principal: Any) -> dict[str, Any]:
                 info = turns.get(run_id) or {}
                 related = [{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level")}
                            for c in row.get("candidates") or [] if c.get("status") == "related"]
-                groups.setdefault((shape.source, shape.text), []).append(_Member(
-                    run_id, query, shape.literals, info.get("conversation_id"), info.get("turn_id"),
-                    str(info.get("question") or ""), related))
+                groups.setdefault((shape.source, shape.key), []).append(_Member(
+                    run_id, query, shape.canonical, info.get("conversation_id"), info.get("turn_id"),
+                    str(info.get("question") or ""), related, shape))
         return groups
 
     groups = await asyncio.to_thread(read)
     items: list[dict[str, Any]] = []
-    for (source, text), members in groups.items():
+    for (source, key), members in groups.items():
         once = list({m.run_id: m for m in members}.values())
         conversations = {m.conversation_id or f"run:{m.run_id}" for m in once}
         if len(once) < REPEAT_RUNS or len(conversations) < REPEAT_CONVERSATIONS:
             continue
-        iid = item_id("shape", f"{source}\n{text}")
-        if iid in dismissed:
+        iid = item_id("shape", f"{source}\n{key}")
+        # 以前按规范文本记的「不是口径」：组里任何一条的规范文本对得上就照样不进队
+        legacy = {item_id("shape", f"{source}\n{m.shape.text}") for m in once if m.shape is not None}
+        if iid in dismissed or legacy & dismissed:
             continue
-        params = varying([m.literals for m in once])
-        related = list({r["caliber"]: r for m in once for r in m.related if r.get("caliber")}.values())
         rep = once[0]
+        shape = rep.shape
+        if shape is None:
+            continue
+        varied = varying([m.literals for m in once])
+        params = [(i, p) for i in varied if i < len(shape.positions) and (p := shape.positions[i]) is not None]
+        related = list({r["caliber"]: r for m in once for r in m.related if r.get("caliber")}.values())
         items.append({
-            "id": iid, "kind": "shape", "source": source, "shape": text, "runs": len(once),
+            "id": iid, "kind": "shape", "source": source, "shape": shape.text, "runs": len(once),
             "conversations": len(conversations), "question": rep.question,
             "turns": [_turn_view(m) for m in once[:TURNS_LIMIT]],
-            "params": [{"index": i, "distinct": len({m.literals[i] for m in once if len(m.literals) > i})}
-                       for i in params],
-            "constants": len(rep.literals) - len(params), "related": related,
+            # index 是代表那条规范文本里的第几个「?」；distinct 按规范形对齐了数
+            "params": [{"index": p, "distinct": len({m.literals[i] for m in once if len(m.literals) > i})}
+                       for i, p in sorted(params, key=lambda ip: ip[1])],
+            "constants": len(shape.literals) - len(params), "related": related,
             "representative": {"run_id": rep.run_id, "query": rep.artifact},
             "peers": [m.run_id for m in once[1:]],
             "frozen_into": _frozen_of([m.run_id for m in once], frozen),
@@ -447,7 +475,7 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name:
         raise CertifyRefused(409, NO_QUERIES)
     kinds = {n: k for n, k in (await session.execute(select(DataSource.name, DataSource.kind))).all()}
 
-    # 同组的别的运行：它们封存的查询里同形状的那一条的字面量
+    # 同组的别的运行：它们封存的查询里同形状（规范形相同，见 Shape.key）的那一条的字面量，按规范形里的先后
     peer_literals: dict[tuple[str, str], list[tuple[str, ...]]] = {}
     peer_sources: list[dict[str, Any]] = []
     peer_runs = [r for r in [await session.get(Run, p) for p in dict.fromkeys(peers) if p != run_id] if r is not None]
@@ -460,7 +488,7 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name:
             except ValueError:
                 continue
             if shape is not None and not shape.problem:
-                peer_literals.setdefault((shape.source, shape.text), []).append(shape.literals)
+                peer_literals.setdefault((shape.source, shape.key), []).append(shape.canonical)
     sealed = await _sealed(run_id)
     shapes: list[Shape] = []
     snaps: list[dict[str, Any]] = []
@@ -485,7 +513,7 @@ async def freeze(session: Any, run_id: str, *, peers: Sequence[str] = (), name:
     metrics: list[dict[str, Any]] = []
     multi = False
     for i, (shape, snap) in enumerate(zip(shapes, snaps)):
-        params = varying([shape.literals, *peer_literals.get((shape.source, shape.text), [])])
+        params = shape.params(varying([shape.canonical, *peer_literals.get((shape.source, shape.key), [])]))
         template, new_fields = template_of(shape, params, kinds.get(shape.source), start=len(fields) + 1)
         fields += new_fields
         nodes.append({"id": nid(i), "type": "tool", "position": {"x": 0, "y": 0},

backend/tests/test_ledger_certify.py

@@ -270,3 +270,80 @@ async def test_freezing_needs_a_matched_successful_run(client, monkeypatch):
     row = await S.explore(monkeypatch, S.ELSEWHERE, text="一共几家店，没有引用。")
     res = await client.post("/api/copilot/from-run", json={"run_id": row.id, "freeze": True})
     assert res.status_code == 409 and res.json()["detail"] == certify.NO_QUERIES
+
+
+# --------------------------------------------------------------------------
+# 确定性重复按规范形分组：条件的先后不算，条件真不同的不算一组
+# --------------------------------------------------------------------------
+
+#: 同一个形状的三种写法：条件换序、AND 两边换位、比较两边换位
+REORDERED = [
+    "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{s}' AND paid_at < '{e}'",
+    "SELECT SUM(amount) AS gmv FROM orders WHERE paid_at < '{e}' AND paid_at >= '{s}' AND status = 'paid'",
+    "SELECT SUM(amount) AS gmv FROM orders WHERE '{s}' <= paid_at AND ('paid' = status AND paid_at < '{e}')",
+]
+
+
+def test_conditions_in_another_order_are_the_same_shape_and_their_literals_line_up():
+    from app.data.sql_semantics import sql_shape
+
+    shapes = [sql_shape(sql.format(s=s, e=e), "sqlite") for sql, (s, e) in zip(REORDERED, WEEKS)]
+    assert all(sh is not None for sh in shapes)
+    assert len({sh.key for sh in shapes}) == 1
+    # 按规范形里的先后对齐：同一个位置是同一个条件里的那个字面量——变了的是两个日期，'paid' 在每一条里都落在同一个位置
+    assert len(certify.varying([sh.literals for sh in shapes])) == 2
+    assert len({sh.literals.index("'paid'") for sh in shapes}) == 1
+    starts = {sh.literals.index(f"'{s}'") for sh, (s, _) in zip(shapes, WEEKS)}
+    assert len(starts) == 1, "起始日期在每一条里都落在同一个位置"
+
+
+@pytest.mark.parametrize("other", [
+    "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND created_at >= '2026-09-01' AND paid_at < '2026-09-08'",
+    "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' OR paid_at >= '2026-09-01' AND paid_at < '2026-09-08'",
+    "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '2026-09-01'",
+    "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at > '2026-09-01' AND paid_at < '2026-09-08'",
+    "SELECT SUM(amount) AS gmv FROM orders WHERE status = 1 AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'",
+    "SELECT SUM(amount) AS gmv FROM orders WHERE status IN ('paid', 'open') AND paid_at >= '2026-09-01' "
+    "AND paid_at < '2026-09-08'",
+    "SELECT COUNT(*) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'",
+], ids=["other-column", "or", "missing-condition", "other-operator", "number-not-string", "in-list", "other-aggregate"])
+def test_really_different_conditions_are_different_shapes(other):
+    from app.data.sql_semantics import sql_shape
+
+    base = sql_shape(REORDERED[0].format(s="2026-09-01", e="2026-09-08"), "sqlite")
+    found = sql_shape(other, "sqlite")
+    assert base is not None and found is not None and found.key != base.key
+
+
+async def test_queue_groups_reordered_conditions_and_freezing_aligns_their_parameters(client, monkeypatch):
+    a, b = await conversation(), await conversation()
+    runs = [await ask(monkeypatch, conv, sql.format(s=s, e=e))
+            for conv, sql, (s, e) in zip((a, a, b), REORDERED, WEEKS)]
+    # 条件真不同的（多筛了一个地区）不进这一组
+    other = await ask(monkeypatch, b, "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND region = 'east' "
+                                      "AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'")
+    [item] = shape_items((await client.get("/api/ledger/certify-queue")).json())
+    assert item["runs"] == 3 and {t["run_id"] for t in item["turns"]} == {r.id for r in runs}
+    assert other.id not in {t["run_id"] for t in item["turns"]}
+    assert len(item["params"]) == 2 and all(p["distinct"] == 3 for p in item["params"]) and item["constants"] == 1
+
+    rep = item["representative"]["run_id"]
+    res = await client.post("/api/copilot/from-run", json={"run_id": rep, "freeze": True, "peers": item["peers"]})
+    assert res.status_code == 200, res.text
+    made = res.json()
+    assert made["params"] == ["p1", "p2"], "条件换了序的几轮照样认得出两个日期是参数"
+    async with SessionLocal() as session:
+        wf = await session.get(Workflow, made["workflow_id"])
+    sql = {n["id"]: n for n in wf.graph["nodes"]}["fetch"]["data"]["config"]["args"]["sql"]
+    assert "'paid'" in sql and "'{{ input.p1 }}'" in sql and "'{{ input.p2 }}'" in sql
+
+
+async def test_a_not_a_caliber_recorded_by_the_old_text_id_still_counts(client, monkeypatch):
+    """以前「不是口径」按规范文本算的 id 记：分组改按规范形之后，组里任何一条的规范文本对得上就照样不进队。"""
+    await three_asks(monkeypatch)
+    [item] = shape_items((await client.get("/api/ledger/certify-queue")).json())
+    old = certify.item_id("shape", f"{S.CURRENT['source']}\n{item['shape']}")
+    assert old != item["id"]
+    res = await client.post("/api/ledger/certify-queue/dismiss", json={"item_id": old, "kind": "shape"})
+    assert res.status_code == 200
+    assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == []