From ae012a8e7cf8b355aebad84245c862c777bb4d1a Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 15:30:12 +0000 Subject: [PATCH] v7 mech-3: scorecard diff scope, gain ledger, stagnation stop table - cloud/scorecard_diff.py: audit_scope (changed files + claimed_fixed, unchanged/baseline-out files out of package), ScorecardGainLedger (memory) + SqliteScorecardGainLedger (SQLite only, PG_DDL text-only), three tables code/production/product kept separately, average reference-only; check_stagnation: two consecutive code gains <0.5 (test_default_unapproved) escalate, no auto pushback; rubric_version change starts a new sequence. - cloud/tests/test_scorecard_mech_diff.py: 24 tests, all passing. - No change to audit_loop_driver.py semantics, #238 auto_apply, #247 correction-pass semantics, or v1.1 contract. (cherry picked from commit 57d44becaf559a074c995249e95bd36924dae13f) --- cloud/scorecard_diff.py | 453 ++++++++++++++++++++++++ cloud/tests/test_scorecard_mech_diff.py | 292 +++++++++++++++ 2 files changed, 745 insertions(+) create mode 100644 cloud/scorecard_diff.py create mode 100644 cloud/tests/test_scorecard_mech_diff.py diff --git a/cloud/scorecard_diff.py b/cloud/scorecard_diff.py new file mode 100644 index 0000000..a88ac23 --- /dev/null +++ b/cloud/scorecard_diff.py @@ -0,0 +1,453 @@ +"""评分卡差分审计与涨幅停表（v7 机制第 3 项，纯函数 + 标准库 only）。 + +只做这一项，不碰其它 v7 机制项。本模块不改动 +``cloud/audit_loop_driver.py`` 的语义，也不改动 +``cloud/audit_scorecard.py``：它是评分卡侧的停表机制，不是 +审计循环本身。审计循环的 pushback 上限 3、重建预算 1、 +版本变更重置思路只作参考，不在本模块里复写或放宽。 + +三张表分开记分，绝不只记平均分 +-------------------------------- +每轮 ledger 条目同时记 ``code`` / ``production`` / ``product`` +三张表各自的分数（任一表未知记 ``None``，不估算）。 +``average_for_reference_only`` 只作参考信息，停表判定只看 +代码表（``code``），绝不用平均分判定——与 audit_scorecard +“平均分绝不能作为达标依据”的规则一致。 + +阈值未拍板 +---------- +``STAGNATION_THRESHOLD`` 的值 0.5 标注 +``source="test_default_unapproved"``（测试默认值、未拍板）， +与 audit_scorecard 的 ``DEFAULT_THRESHOLDS`` 口径一致。 +生产阈值须用户拍板后由调用方显式传入 ``threshold`` 才能替换； +本模块绝不把测试初值写成生产值，也不替用户决定 B/C 与阈值。 + +与已定语义的关系（本模块不改它们） +---------------------------------- +- #238 ``evaluate_feedback`` 的 ``auto_apply`` 保持 False； + 本模块没有 auto_apply 路径。 +- #247 纠正为 pass 本身不解封；本模块不做解封、不做 + re-audit、不碰门禁状态，只返回停表判定。 +- v1.1 契约不变；本模块不增维、不改 7 维定义。 +- 停表触发时返回 ``escalate`` 语义，不再自动推回： + 结果里 ``auto_pushback`` 恒为 False，且 + ``escalation_grants_user_acceptance`` 恒为 False + （escalate 从不 tick 用户验收，与 audit_loop_driver 一致）。 + +(a) 差分范围输入：``audit_scope(baseline_snapshot, current_snapshot, +claimed_fixed)`` +---------------------------------------------------------------- +快照形式（二选一，两种都接受）： +- 文件路径集合：``set`` / ``list`` / ``tuple``，元素为路径字符串。 + 此时变更 = 对称差（新增 + 删除）；交集 = 未变更。 +- 快照 dict：键为路径，值为内容指纹（hash/内容字符串），或 + ``{"files": <路径集合或路径->指纹 dict>}`` 形式。 + 此时变更 = 新增 + 删除 + 同路径指纹不同（修改）；指纹相同 + 的路径 = 未变更。 + +审计范围 = 变更文件 + 调用方声称修好的项（``claimed_fixed``）。 +未变更文件进 ``out_of_scope``，不进审计包；基线外（即不在 +变更集内、也不在声称修好集内）的文件一律不进包。 +``claimed_fixed`` 是项 id（检查名/维 id 等字符串），不是文件 +路径；它与文件集分开返回，调用方按项复核，不与文件混算。 + +(b) 涨幅 ledger：``ScorecardGainLedger``（内存）与 +``SqliteScorecardGainLedger``（SQLite） +------------------------------------------------ +- 内存 ledger：append-only 列表，``append()`` 存规范化后的 + 条目副本，``rounds()`` 返回副本（调用方改返回结果不影响 + 已存条目）。模块级函数 ``append_round`` / ``read_rounds`` + 接受 ledger 对象或普通 list，方便测试与调用方复用。 +- SQLite ledger：同一列名、同一追加语义的 SQLite 表 + ``scorecard_gain_rounds``（见 ``SQLITE_DDL``），用标准库 + ``sqlite3``、``?`` 占位符。``PG_DDL`` 是与之列对应的 + Postgres 建表语句文本，仅作仓库 DB 模式兼容写法： + 真实 PG 在本项中不接通、不执行、不验证，只用 + SQLite（内存或文件）测试并明说。 +- 条目校验：``rubric_version`` 必须为非空字符串，否则 + ``ValueError``；三张表分数各为 ``None`` 或 0-10 的数， + 超范围或非数值 likewise ``ValueError``；bool 不算数值。 + +(c) 停表判定：``check_stagnation(history)`` +-------------------------------------------- +- 只看 trailing 同 rubric_version 序列：从最新一轮往回取 + 与最新轮 ``rubric_version`` 相同的连续轮次；改标尺即新 + 序列，旧标尺的轮次不连算。 +- 该序列内相邻两轮代码表分数之差为一轮涨幅；分数为 1 位 + 小数（与 audit_scorecard 的 _round1 一致），涨幅按 + 1 位小数舍入后再与阈值比较，避免 8.8-8.3 这类二进制 + 浮点残差把恰好 0.5 误判为 <0.5。 +- 涨幅 < 阈值（含负涨幅，即代码表下滑）记为小涨幅； + 涨幅 >= 阈值，或涨幅未知（任一端分数为 None），打断 + 连续计数。连续两轮小涨幅（即该序列内最近两次涨幅 + 均为小涨幅，至少需要 3 轮）即触发停表。 +- 触发返回 ``{"decision": "escalate", "reason": "stagnation", ...}``； + 未触发返回 ``{"decision": "continue", ...}``。两种结果都 + 不含 pushback 动作；触发时 ``auto_pushback`` 为 False。 +""" + +from __future__ import annotations + +import sqlite3 + +TABLES = ("code", "production", "product") + +# 停表阈值：测试默认值、未拍板。生产阈值须用户拍板后显式传入。 +STAGNATION_THRESHOLD = {"value": 0.5, "source": "test_default_unapproved"} + +# 触发所需连续小涨幅轮数（2 次小涨幅 = 序列内至少 3 轮）。 +STAGNATION_CONSECUTIVE_GAINS = 2 + +SQLITE_DDL = ( + "CREATE TABLE IF NOT EXISTS scorecard_gain_rounds (" + "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "rubric_version TEXT NOT NULL, " + "code REAL, " + "production REAL, " + "product REAL, " + "average_for_reference_only REAL" + ")" +) + +# 与 SQLITE_DDL 列对应的 Postgres 建表文本。仅作模式兼容写法， +# 本项不接通真实 PG、不执行、不验证；测试只覆盖 SQLite。 +PG_DDL = ( + "CREATE TABLE IF NOT EXISTS scorecard_gain_rounds (" + "id SERIAL PRIMARY KEY, " + "rubric_version TEXT NOT NULL, " + "code DOUBLE PRECISION, " + "production DOUBLE PRECISION, " + "product DOUBLE PRECISION, " + "average_for_reference_only DOUBLE PRECISION" + ")" +) + + +# ------------------------------------------------------------------ +# (a) 差分范围输入 +# ------------------------------------------------------------------ + +def _fingerprint(value): + """快照 dict 值的规范指纹：hash/sha256/content 字段优先，否则原值。""" + if isinstance(value, dict): + for key in ("sha256", "hash", "content"): + if value.get(key) is not None: + return value[key] + return repr(sorted(value.items())) + return value + + +def _snapshot_mapping(snapshot): + """把快照归一化为 {路径: 指纹或 None}。 + + 路径集合形式：指纹为 None（只比存在性）。 + 快照 dict 形式：键为路径，值为指纹；{"files": ...} 形式先解包。 + None 快照视为空快照。 + """ + if snapshot is None: + return {} + if isinstance(snapshot, dict): + inner = snapshot.get("files", snapshot) if "files" in snapshot else snapshot + if isinstance(inner, dict): + return {str(path): _fingerprint(value) for path, value in inner.items()} + if isinstance(inner, (set, list, tuple)): + return {str(path): None for path in inner} + raise TypeError("snapshot files must be a mapping or a path collection") + if isinstance(snapshot, (set, list, tuple)): + return {str(path): None for path in snapshot} + raise TypeError("snapshot must be a path collection, a mapping, or None") + + +def _normalize_claimed(claimed_fixed): + """声称修好的项：None -> []；单个字符串 -> [它]；否则去重排序。""" + if claimed_fixed is None: + return [] + if isinstance(claimed_fixed, str): + return [claimed_fixed] + return sorted({str(item) for item in claimed_fixed}) + + +def audit_scope(baseline_snapshot, current_snapshot, claimed_fixed=None): + """算出差分审计范围。 + + 返回 {"in_scope": {"files": [...], "claimed_fixed": [...]}, + "out_of_scope": {"files": [...]}}。 + in_scope.files = 变更文件（新增 + 删除 + 修改）； + out_of_scope.files = 两快照中都出现且指纹相同（路径集合 + 形式下即交集）的未变更文件；claimed_fixed 原样进包， + 与文件集分开，不因文件未变更而被排除。 + """ + baseline = _snapshot_mapping(baseline_snapshot) + current = _snapshot_mapping(current_snapshot) + # 以存在性为准：只在一侧出现的路径必为变更；路径集合形式下 + # 指纹均为 None，不能只比指纹（缺失侧 get 缺省也是 None， + # 会把新增/删除误判为未变更），故先判存在性再比指纹。 + changed = sorted( + path + for path in set(baseline) | set(current) + if path not in baseline or path not in current or baseline[path] != current[path] + ) + unchanged = sorted( + path + for path in set(baseline) & set(current) + if baseline[path] == current[path] + ) + return { + "in_scope": { + "files": changed, + "claimed_fixed": _normalize_claimed(claimed_fixed), + }, + "out_of_scope": {"files": unchanged}, + } + + +# ------------------------------------------------------------------ +# (b) 涨幅 ledger：条目规范化 +# ------------------------------------------------------------------ + +def _is_number(value): + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def normalize_round(rubric_version, code=None, production=None, product=None, entry=None): + """规范化一轮 ledger 条目（返回新 dict，不修改入参）。 + + 接受两种调用：显式三表分数，或 entry dict（含 flat 的 + code/production/product，或嵌套的 scores dict）。 + 平均分只在三表全已知时算出，且只作参考，永不用于停表。 + """ + if entry is not None: + if not isinstance(entry, dict): + raise TypeError("entry must be a dict") + scores = entry.get("scores") if isinstance(entry.get("scores"), dict) else entry + rubric_version = entry.get("rubric_version", rubric_version) + code = scores.get("code", code) if code is None else code + production = scores.get("production", production) if production is None else production + product = scores.get("product", product) if product is None else product + if not isinstance(rubric_version, str) or not rubric_version: + raise ValueError("rubric_version must be a non-empty string") + values = {} + for name, value in (("code", code), ("production", production), ("product", product)): + if value is None: + values[name] = None + elif _is_number(value) and 0 <= float(value) <= 10: + values[name] = round(float(value) + 1e-9, 1) + else: + raise ValueError("%s must be None or a number in 0..10, got %r" % (name, value)) + known = [values[name] for name in TABLES] + average = round(sum(known) / len(known) + 1e-9, 1) if all(v is not None for v in known) else None + return { + "rubric_version": rubric_version, + "code": values["code"], + "production": values["production"], + "product": values["product"], + "average_for_reference_only": average, + "average_is_not_pass_criterion": True, + } + + +class ScorecardGainLedger: + """内存涨幅 ledger：append-only，存条目副本，读出副本。""" + + def __init__(self): + self._rounds = [] + + def append(self, rubric_version=None, code=None, production=None, product=None, entry=None): + normalized = normalize_round(rubric_version, code, production, product, entry) + self._rounds.append(normalized) + return dict(normalized) + + def rounds(self): + return [dict(entry) for entry in self._rounds] + + # history() 是 rounds() 的别名，供 check_stagnation 直接接受 ledger。 + def history(self): + return self.rounds() + + +def append_round(ledger, rubric_version=None, code=None, production=None, product=None, entry=None): + """向 ledger 追加一轮：ledger 对象（有 .append）或普通 list 均可。""" + normalized = normalize_round(rubric_version, code, production, product, entry) + if hasattr(ledger, "append") and not isinstance(ledger, list): + # ledger 对象自身的 append 已做规范化；为避免双重规范化 + # 带来的舍入差异，这里直接传规范化后的 entry。 + ledger.append(entry=normalized) + return dict(normalized) + ledger.append(normalized) + return dict(normalized) + + +def read_rounds(ledger): + """读出 ledger 全部轮次副本：ledger 对象（有 .rounds）或 list 均可。""" + if hasattr(ledger, "rounds"): + return [dict(entry) for entry in ledger.rounds()] + return [dict(entry) for entry in ledger] + + +class SqliteScorecardGainLedger: + """SQLite 涨幅 ledger：与内存 ledger 同列、同追加语义。 + + db_path 为 ":memory:"（默认）或文件路径；传入 sqlite3 + 连接亦可（此时 close() 不关闭调用方连接）。真实 PG + 不接通：PG_DDL 只是模式兼容文本，不在本类中执行。 + """ + + def __init__(self, db_path=":memory:", connection=None): + if connection is not None: + self._conn = connection + self._owns_conn = False + else: + self._conn = sqlite3.connect(str(db_path)) + self._conn.row_factory = sqlite3.Row + self._owns_conn = True + self._conn.execute(SQLITE_DDL) + self._conn.commit() + + def append(self, rubric_version=None, code=None, production=None, product=None, entry=None): + normalized = normalize_round(rubric_version, code, production, product, entry) + self._conn.execute( + "INSERT INTO scorecard_gain_rounds" + " (rubric_version, code, production, product, average_for_reference_only)" + " VALUES (?, ?, ?, ?, ?)", + (normalized["rubric_version"], normalized["code"], normalized["production"], + normalized["product"], normalized["average_for_reference_only"]), + ) + self._conn.commit() + return dict(normalized) + + def rounds(self): + rows = self._conn.execute( + "SELECT rubric_version, code, production, product," + " average_for_reference_only FROM scorecard_gain_rounds ORDER BY id" + ).fetchall() + out = [] + for row in rows: + mapping = dict(row) if isinstance(row, sqlite3.Row) else dict(row) + entry = { + "rubric_version": mapping["rubric_version"], + "code": mapping["code"], + "production": mapping["production"], + "product": mapping["product"], + "average_for_reference_only": mapping["average_for_reference_only"], + "average_is_not_pass_criterion": True, + } + out.append(entry) + return out + + def history(self): + return self.rounds() + + def close(self): + if self._owns_conn: + self._conn.close() + + +# ------------------------------------------------------------------ +# (c) 停表判定 +# ------------------------------------------------------------------ + +def _history_entries(history): + """接受 list/tuple 或 ledger 对象（有 .rounds/.history），返回条目 list。""" + if history is None: + return [] + if hasattr(history, "rounds"): + return list(history.rounds()) + if hasattr(history, "history"): + return list(history.history()) + return list(history) + + +def _code_score(entry): + """取一轮的代码表分数：flat 的 code，或嵌套 scores.code；缺失为 None。""" + if not isinstance(entry, dict): + return None + if entry.get("code") is not None: + return entry["code"] + scores = entry.get("scores") + if isinstance(scores, dict): + return scores.get("code") + return None + + +def _trailing_run(entries): + """最新 rubric_version 的 trailing 连续序列（改标尺即新序列）。""" + if not entries: + return [] + latest_version = entries[-1].get("rubric_version") if isinstance(entries[-1], dict) else None + run = [] + for entry in reversed(entries): + version = entry.get("rubric_version") if isinstance(entry, dict) else None + if version != latest_version: + break + run.append(entry) + run.reverse() + return run + + +def code_gains(history): + """trailing 同标尺序列内相邻两轮的代码表涨幅 list。 + + 每项为 1 位小数舍入后的涨幅，或 None（任一端分数未知）。 + 全序列的涨幅不返回：旧标尺轮次不连算。 + """ + run = _trailing_run(_history_entries(history)) + scores = [_code_score(entry) for entry in run] + gains = [] + for prev_score, cur_score in zip(scores, scores[1:]): + if prev_score is None or cur_score is None: + gains.append(None) + elif not _is_number(prev_score) or not _is_number(cur_score): + gains.append(None) + else: + gains.append(round(float(cur_score) - float(prev_score) + 1e-9, 1)) + return gains + + +def _resolve_threshold(threshold): + """阈值解析：None -> 测试默认值；数字 -> 该值（来源仍标测试默认， + 除非调用方显式传 {"value":..., "source": "user_approved"}）； + dict -> 取 value/source。""" + if threshold is None: + return STAGNATION_THRESHOLD["value"], STAGNATION_THRESHOLD["source"] + if isinstance(threshold, dict): + return threshold.get("value", STAGNATION_THRESHOLD["value"]), threshold.get( + "source", STAGNATION_THRESHOLD["source"]) + return threshold, STAGNATION_THRESHOLD["source"] + + +def check_stagnation(history, threshold=None): + """停表判定：trailing 同标尺序列内连续两轮代码表涨幅 < 阈值即停。 + + 触发返回 decision="escalate"、reason="stagnation"，且 + auto_pushback=False（不再自动推回）； + 未触发返回 decision="continue"。两种结果的 + escalation_grants_user_acceptance 均为 False。 + """ + threshold_value, threshold_source = _resolve_threshold(threshold) + entries = _history_entries(history) + run = _trailing_run(entries) + gains = code_gains(entries) + consecutive = 0 + for gain in reversed(gains): + if gain is not None and gain < threshold_value: + consecutive += 1 + else: + break + latest_version = run[-1].get("rubric_version") if run else None + base = { + "rubric_version": latest_version, + "run_rounds": len(run), + "code_gains": gains, + "consecutive_small_gains": consecutive, + "threshold": threshold_value, + "threshold_source": threshold_source, + "auto_pushback": False, + "escalation_grants_user_acceptance": False, + } + if consecutive >= STAGNATION_CONSECUTIVE_GAINS: + result = {"decision": "escalate", "reason": "stagnation"} + result.update(base) + return result + result = {"decision": "continue", "reason": "not_stagnant"} + result.update(base) + return result diff --git a/cloud/tests/test_scorecard_mech_diff.py b/cloud/tests/test_scorecard_mech_diff.py new file mode 100644 index 0000000..674a01d --- /dev/null +++ b/cloud/tests/test_scorecard_mech_diff.py @@ -0,0 +1,292 @@ +"""v7 机制第 3 项：差分审计与涨幅停表定向测试。 + +钉住的规则（改动前先改 cloud/scorecard_diff.py 的 docstring）： +- 审计范围 = 变更文件 + 声称修好的项；未变更/基线外文件不进包。 +- 每轮记 code/production/product 三张表各自的分数，绝不只记平均分； + 停表只看代码表，不看平均分。 +- 连续两轮代码表涨幅 <0.5（阈值测试默认值、未拍板）即停表， + 返回 escalate、不再自动推回；涨幅 >=0.5 或只有一轮 <0.5 不触发。 +- 改标尺即新序列：不同 rubric_version 的轮次不连算。 +- 真实 PG 不接通：SQLite/内存测试并明说；PG_DDL 只是模式兼容文本。 +- 本模块不改 audit_loop_driver 语义、不改 #238 auto_apply、 + 不改 #247 纠正为 pass 不解封、不改 v1.1 契约。 +""" + +import os +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.scorecard_diff import ( + PG_DDL, + SQLITE_DDL, + STAGNATION_THRESHOLD, + ScorecardGainLedger, + SqliteScorecardGainLedger, + append_round, + audit_scope, + check_stagnation, + code_gains, + read_rounds, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.scorecard_diff import ( + PG_DDL, + SQLITE_DDL, + STAGNATION_THRESHOLD, + ScorecardGainLedger, + SqliteScorecardGainLedger, + append_round, + audit_scope, + check_stagnation, + code_gains, + read_rounds, + ) + + +def _round(version, code, production=None, product=None): + return {"rubric_version": version, "code": code, + "production": production, "product": product} + + +class AuditScopeTest(unittest.TestCase): + def test_path_set_changed_in_scope_unchanged_out(self): + scope = audit_scope( + {"cloud/server.py", "cloud/foo.py"}, + {"cloud/server.py", "cloud/bar.py"}, + claimed_fixed=[], + ) + self.assertEqual(scope["in_scope"]["files"], ["cloud/bar.py", "cloud/foo.py"]) + self.assertEqual(scope["out_of_scope"]["files"], ["cloud/server.py"]) + self.assertNotIn("cloud/server.py", scope["in_scope"]["files"]) + + def test_dict_snapshot_modified_by_hash_in_scope_same_hash_out(self): + scope = audit_scope( + {"a.py": "hash1", "b.py": "hash2"}, + {"a.py": "hash1", "b.py": "hash9"}, + ) + self.assertEqual(scope["in_scope"]["files"], ["b.py"]) + self.assertEqual(scope["out_of_scope"]["files"], ["a.py"]) + + def test_added_and_removed_are_changed(self): + scope = audit_scope({"old.py": "h1"}, {"new.py": "h2"}) + self.assertEqual(scope["in_scope"]["files"], ["new.py", "old.py"]) + self.assertEqual(scope["out_of_scope"]["files"], []) + + def test_baseline_outside_files_excluded_from_package(self): + # 基线外文件：未变更、也不在声称修好集内 -> 不进审计包， + # 只出现在 out_of_scope，不出现在 in_scope 的任何位置。 + scope = audit_scope( + {"cloud/server.py": "s1", "cloud/untouched.py": "u1", "cloud/other.py": "o1"}, + {"cloud/server.py": "s2", "cloud/untouched.py": "u1", "cloud/other.py": "o1"}, + claimed_fixed=["check_changed"], + ) + package = scope["in_scope"]["files"] + scope["in_scope"]["claimed_fixed"] + self.assertNotIn("cloud/untouched.py", package) + self.assertNotIn("cloud/other.py", package) + self.assertIn("cloud/untouched.py", scope["out_of_scope"]["files"]) + self.assertIn("cloud/other.py", scope["out_of_scope"]["files"]) + self.assertEqual(scope["in_scope"]["files"], ["cloud/server.py"]) + + def test_claimed_fixed_in_scope_even_when_file_unchanged(self): + scope = audit_scope( + {"a.py": "h1"}, {"a.py": "h1"}, + claimed_fixed=["check_x", "check_x", "check_a"], + ) + self.assertEqual(scope["in_scope"]["files"], []) + self.assertEqual(scope["in_scope"]["claimed_fixed"], ["check_a", "check_x"]) + self.assertEqual(scope["out_of_scope"]["files"], ["a.py"]) + + def test_files_key_form_and_none_claimed(self): + scope = audit_scope({"files": ["a.py"]}, {"files": ["a.py", "b.py"]}) + self.assertEqual(scope["in_scope"]["files"], ["b.py"]) + self.assertEqual(scope["in_scope"]["claimed_fixed"], []) + self.assertEqual(scope["out_of_scope"]["files"], ["a.py"]) + + def test_empty_baseline_all_current_changed(self): + scope = audit_scope(None, {"a.py", "b.py"}) + self.assertEqual(scope["in_scope"]["files"], ["a.py", "b.py"]) + self.assertEqual(scope["out_of_scope"]["files"], []) + + +class GainLedgerTest(unittest.TestCase): + def test_memory_ledger_keeps_three_tables_not_only_average(self): + ledger = ScorecardGainLedger() + entry = append_round(ledger, "v1", code=8.0, production=7.0, product=9.0) + self.assertEqual(entry["code"], 8.0) + self.assertEqual(entry["production"], 7.0) + self.assertEqual(entry["product"], 9.0) + self.assertEqual(entry["average_for_reference_only"], 8.0) + self.assertTrue(entry["average_is_not_pass_criterion"]) + rounds = read_rounds(ledger) + self.assertEqual(len(rounds), 1) + self.assertEqual(rounds[0]["code"], 8.0) + self.assertEqual(rounds[0]["production"], 7.0) + self.assertEqual(rounds[0]["product"], 9.0) + + def test_average_none_when_any_table_unknown(self): + ledger = ScorecardGainLedger() + append_round(ledger, "v1", code=8.0, production=None, product=9.0) + self.assertIsNone(read_rounds(ledger)[0]["average_for_reference_only"]) + self.assertEqual(read_rounds(ledger)[0]["code"], 8.0) + + def test_read_returns_copies_append_stores_copies(self): + ledger = ScorecardGainLedger() + entry = {"rubric_version": "v1", "code": 8.0, "production": 8.0, "product": 8.0} + append_round(ledger, entry=entry) + entry["code"] = 0.0 + self.assertEqual(read_rounds(ledger)[0]["code"], 8.0) + read_rounds(ledger)[0]["code"] = 0.0 + self.assertEqual(read_rounds(ledger)[0]["code"], 8.0) + + def test_plain_list_ledger(self): + ledger = [] + append_round(ledger, "v1", code=8.0, production=8.0, product=8.0) + self.assertEqual(len(read_rounds(ledger)), 1) + self.assertEqual(read_rounds(ledger)[0]["rubric_version"], "v1") + + def test_invalid_round_rejected(self): + ledger = ScorecardGainLedger() + with self.assertRaises(ValueError): + append_round(ledger, "", code=8.0, production=8.0, product=8.0) + with self.assertRaises(ValueError): + append_round(ledger, "v1", code=11.0, production=8.0, product=8.0) + with self.assertRaises(ValueError): + append_round(ledger, "v1", code=True, production=8.0, product=8.0) + self.assertEqual(read_rounds(ledger), []) + + def test_sqlite_memory_roundtrip_and_ddl(self): + self.assertIn("scorecard_gain_rounds", SQLITE_DDL) + self.assertIn("rubric_version", SQLITE_DDL) + self.assertIn("code", SQLITE_DDL) + ledger = SqliteScorecardGainLedger() + try: + append_round(ledger, "v1", code=8.0, production=7.5, product=9.0) + append_round(ledger, "v1", code=8.3, production=7.5, product=9.0) + rounds = read_rounds(ledger) + self.assertEqual([r["code"] for r in rounds], [8.0, 8.3]) + self.assertEqual(rounds[0]["production"], 7.5) + self.assertEqual(rounds[1]["product"], 9.0) + finally: + ledger.close() + + def test_sqlite_file_persists_across_reopen(self): + tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + tmp.close() + try: + ledger = SqliteScorecardGainLedger(db_path=tmp.name) + try: + append_round(ledger, "v1", code=8.0, production=8.0, product=8.0) + finally: + ledger.close() + reopened = SqliteScorecardGainLedger(db_path=tmp.name) + try: + self.assertEqual(len(read_rounds(reopened)), 1) + self.assertEqual(read_rounds(reopened)[0]["code"], 8.0) + finally: + reopened.close() + finally: + os.unlink(tmp.name) + + def test_pg_ddl_is_text_only_and_mirrors_columns(self): + # PG 不接通：只断言 PG_DDL 文本与 SQLite 列名对应，不连真实 PG。 + for column in ("rubric_version", "code", "production", "product", + "average_for_reference_only"): + self.assertIn(column, PG_DDL) + self.assertIn("scorecard_gain_rounds", PG_DDL) + conn = sqlite3.connect(":memory:") + try: + conn.execute(SQLITE_DDL) + finally: + conn.close() + + +class StagnationTest(unittest.TestCase): + def test_threshold_is_test_default_unapproved(self): + self.assertEqual(STAGNATION_THRESHOLD["value"], 0.5) + self.assertEqual(STAGNATION_THRESHOLD["source"], "test_default_unapproved") + + def test_two_small_gains_trigger_escalate_no_pushback(self): + history = [_round("v1", 8.0), _round("v1", 8.3), _round("v1", 8.5)] + self.assertEqual(code_gains(history), [0.3, 0.2]) + result = check_stagnation(history) + self.assertEqual(result["decision"], "escalate") + self.assertEqual(result["reason"], "stagnation") + self.assertFalse(result["auto_pushback"]) + self.assertFalse(result["escalation_grants_user_acceptance"]) + self.assertEqual(result["consecutive_small_gains"], 2) + self.assertEqual(result["rubric_version"], "v1") + + def test_gain_at_or_above_threshold_does_not_trigger(self): + # 0.6 大涨幅打断连续计数：最近两次涨幅为 0.6、0.2，只一轮小涨幅。 + result = check_stagnation([_round("v1", 8.0), _round("v1", 8.6), _round("v1", 8.8)]) + self.assertEqual(result["decision"], "continue") + self.assertEqual(result["consecutive_small_gains"], 1) + # 恰好 0.5 不算 <0.5：两次 0.5 涨幅也不触发。 + result = check_stagnation([_round("v1", 8.0), _round("v1", 8.5), _round("v1", 9.0)]) + self.assertEqual(result["decision"], "continue") + self.assertEqual(code_gains([_round("v1", 8.0), _round("v1", 8.5), _round("v1", 9.0)]), + [0.5, 0.5]) + + def test_single_small_gain_does_not_trigger(self): + result = check_stagnation([_round("v1", 8.0), _round("v1", 8.3)]) + self.assertEqual(result["decision"], "continue") + self.assertEqual(result["consecutive_small_gains"], 1) + self.assertEqual(check_stagnation([_round("v1", 8.0)])["decision"], "continue") + self.assertEqual(check_stagnation([])["decision"], "continue") + + def test_rubric_version_change_starts_new_sequence(self): + # 旧标尺 v1 有两次小涨幅，但最新轮已改标尺为 v2：v2 序列内 + # 只有一轮 0.2 涨幅，不连算旧序列，不触发。 + history = [_round("v1", 8.0), _round("v1", 8.3), _round("v1", 8.5), + _round("v2", 8.0), _round("v2", 8.2)] + self.assertEqual(code_gains(history), [0.2]) + result = check_stagnation(history) + self.assertEqual(result["decision"], "continue") + self.assertEqual(result["rubric_version"], "v2") + # v2 序列内攒够两次小涨幅后才触发。 + history.append(_round("v2", 8.4)) + self.assertEqual(check_stagnation(history)["decision"], "escalate") + + def test_unknown_code_score_breaks_streak(self): + history = [_round("v1", 8.0), _round("v1", 8.3), _round("v1", None)] + self.assertEqual(code_gains(history), [0.3, None]) + self.assertEqual(check_stagnation(history)["decision"], "continue") + + def test_negative_gain_counts_as_small_gain(self): + # 代码表下滑也是未达阈值：连续两次下滑触发 escalate，不是 pushback。 + result = check_stagnation([_round("v1", 9.0), _round("v1", 8.8), _round("v1", 8.6)]) + self.assertEqual(code_gains([_round("v1", 9.0), _round("v1", 8.8), _round("v1", 8.6)]), + [-0.2, -0.2]) + self.assertEqual(result["decision"], "escalate") + self.assertFalse(result["auto_pushback"]) + + def test_nested_scores_form_and_ledger_input(self): + history = [ + {"rubric_version": "v1", "scores": {"code": 8.0, "production": 8.0, "product": 8.0}}, + {"rubric_version": "v1", "scores": {"code": 8.2, "production": 8.0, "product": 8.0}}, + {"rubric_version": "v1", "scores": {"code": 8.4, "production": 8.0, "product": 8.0}}, + ] + self.assertEqual(check_stagnation(history)["decision"], "escalate") + ledger = ScorecardGainLedger() + append_round(ledger, "v1", code=8.0, production=8.0, product=8.0) + append_round(ledger, "v1", code=8.2, production=8.0, product=8.0) + append_round(ledger, "v1", code=8.4, production=8.0, product=8.0) + self.assertEqual(check_stagnation(ledger)["decision"], "escalate") + + def test_average_gain_does_not_drive_stagnation(self): + # 三表平均涨幅大、代码表涨幅小：停表只看代码表，仍触发。 + history = [ + _round("v1", 8.0, production=5.0, product=5.0), + _round("v1", 8.2, production=9.0, product=9.0), + _round("v1", 8.4, production=9.0, product=9.0), + ] + self.assertEqual(check_stagnation(history)["decision"], "escalate") + + +if __name__ == "__main__": + unittest.main()