"""T11 改进治理：旧/新对照、独立审核与发布。 依据：docs/v2/06-validation.md §5/§7、14-experience-loop.md §8、 03-detailed-design.md §9.4；BR-15/16/40、E-11/24、AC-15/16/34/37/41。 四条不可让渡的规则： 1. **预测先锁定，标签后揭示**：任何一条在锁定前就拿到目标标签的样本都算泄漏， 该样本作废（AC-15）；保留集来源不得进入开发集（AC-36）。 2. **没有对照不发布**：缺证据、非独立评测、退化、结论不明，任一项即为阻塞（AC-34）。 3. **自批不算审核**：提出者不能审核自己的候选（AC-41）。 4. **发布引用不可变版本**：发布只指向既有发布结果；旧版本始终可回放。 """ from __future__ import annotations from contracts.improvement import ContrastMetrics, ContrastReport, ImprovementCandidate from services import assistant_store as store KIND = "improvement_candidate" class LeakageDetected(RuntimeError): """对照样本发生泄漏（E-11）：样本作废，不得据此得出改进结论。""" class NotValidated(RuntimeError): """未通过对照/审核（E-24）。""" def _rate(numerator: int, denominator: int) -> float | None: return None if denominator == 0 else round(numerator / denominator, 4) def evaluate_contrast(samples, baseline_ref: dict, candidate_ref: dict, *, dataset_ref: str, split_ref: str, reviewer_id: str | None, independent: bool) -> dict: """按冻结样本算旧/新对照（AC-15/37）。 样本字段：id、split（dev/holdout）、predicted_at、label_revealed_at、 observed、baseline_pass、candidate_pass、source_in_dev。 泄漏样本（锁定前已见标签，或保留集来源进了开发集）一律作废并单列。 """ premature = [item["id"] for item in samples if item.get("label_revealed_at") and item.get("predicted_at") and str(item["label_revealed_at"]) < str(item["predicted_at"])] if premature: raise LeakageDetected( f"以下样本在预测锁定前已读取目标标签，评测作废：{sorted(premature)}") contaminated = sorted(item["id"] for item in samples if item.get("split") == "holdout" and item.get("source_in_dev")) usable = [item for item in samples if item["id"] not in set(contaminated)] def arm(key: str, label: str) -> ContrastMetrics: observed = [item for item in usable if item.get("observed")] passed = [item for item in observed if item.get(key) is True] denominator = len(observed) return ContrastMetrics( label=label, numerator=len(passed), observable_denominator=denominator, unobserved=len(usable) - denominator, coverage=round(denominator / len(usable), 4) if (usable and denominator) else None, rate=_rate(len(passed), denominator)) baseline = arm("baseline_pass", "现有助手") candidate = arm("candidate_pass", "主审 + 历史经验（候选）") if not independent: # 同模型自评不算独立评测：只能记"结论不明"，不得声称改进（AC-41） verdict = "inconclusive" elif baseline.observable_denominator == 0 or candidate.observable_denominator == 0: verdict = "inconclusive" elif (candidate.rate or 0) > (baseline.rate or 0): verdict = "improved" elif (candidate.rate or 0) < (baseline.rate or 0): verdict = "regressed" else: verdict = "unchanged" report = ContrastReport( id=store.uid("contrast"), baseline_ref=baseline_ref, candidate_ref=candidate_ref, dataset_ref=dataset_ref, split_ref=split_ref, baseline_metrics=[baseline], candidate_metrics=[candidate], sample_size=len(usable), contaminated=contaminated, independent=independent, verdict=verdict, reviewer_id=reviewer_id, created_at=store.now()) return report.model_dump(mode="json") def _candidate(db, candidate_id: str) -> dict: return store.get(db, KIND, candidate_id) def propose(db, project_id: str, body: dict, actor: str, data_domain: str = "demo") -> dict: """提出候选。必须写明改什么、依据哪些运行/结果，以及预期的发布版本引用。""" if not body.get("change_summary"): raise ValueError("改进候选必须说明改了什么，不能只有标题") if not body.get("source_run_ids") and not body.get("outcome_ids"): raise ValueError("改进候选必须引用来源运行或会议结果，凭空提出的改动不进入审核") candidate = ImprovementCandidate( id=store.uid("improvement"), project_id=project_id, data_domain=data_domain, revision=1, target=body.get("target") or "knowledge", target_ref=body["target_ref"], source_run_ids=list(body.get("source_run_ids") or ()), outcome_ids=list(body.get("outcome_ids") or ()), change_summary=body["change_summary"], proposed_revision_ref=body["proposed_revision_ref"], status="draft", proposed_by=actor, created_at=store.now()) saved = candidate.model_dump(mode="json") store.put(db, KIND, saved) store.event(db, project_id, "improvement.proposed", actor, {"candidate_id": saved["id"], "target": saved["target"]}, None) return saved def _save(db, body: dict, actor: str, action: str) -> dict: body["revision"] = int(body.get("revision") or 1) + 1 saved = ImprovementCandidate(**body).model_dump(mode="json") store.put(db, KIND, saved, expected=None) store.event(db, saved["project_id"], f"improvement.{action}", actor, {"candidate_id": saved["id"], "status": saved["status"]}, None) return saved def transition(db, candidate_id: str, action: str, actor: str, expected_revision: int | None = None, *, evaluation: dict | None = None, contrast: dict | None = None, released_ref: dict | None = None, publisher=None, reason: str | None = None) -> dict: """候选状态机。发布委托给注入的 publisher，并要求引用既有发布结果。""" body = dict(_candidate(db, candidate_id)) if expected_revision is not None and body.get("revision") != expected_revision: raise store.Conflict("候选已有新版本，请刷新后重试") if action == "evaluate": if evaluation is not None: # 泄漏与指标一律在服务端判定：不接受客户端直接交来的"成品报告" contrast = evaluate_contrast( evaluation.get("samples") or (), evaluation["baseline_ref"], evaluation["candidate_ref"], dataset_ref=evaluation["dataset_ref"], split_ref=evaluation["split_ref"], reviewer_id=evaluation.get("reviewer_id"), independent=bool(evaluation.get("independent"))) elif contrast is not None: contrast = ContrastReport(**contrast).model_dump(mode="json") else: raise ValueError("评测必须提交样本与版本引用，不能只改状态") body["contrast"] = contrast body["evaluation_ref"] = {"id": contrast["id"], "version": "1"} if contrast.get("contaminated"): body["status"] = "evaluating" # 有泄漏：留在评测中，不得进待审核 body["rejection_reason"] = "存在泄漏样本，已作废；请修正数据划分后重测" elif contrast["verdict"] in ("improved", "unchanged") and contrast["independent"]: body["status"] = "ready" else: body["status"] = "evaluating" return _save(db, body, actor, action) if action == "submit": if body.get("contrast") is None: raise NotValidated("没有对照证据的候选不得提交审核") body["status"] = "in_review" return _save(db, body, actor, action) if action == "approve": ImprovementCandidate(**{**body, "status": "approved", "reviewer_id": actor}).must_not_self_approve(actor) if body.get("contrast") is None: raise NotValidated("没有对照证据的候选不得通过审核") body["status"] = "approved" body["reviewer_id"] = actor return _save(db, body, actor, action) if action == "reject": body["status"] = "rejected" body["rejection_reason"] = reason or "审核未通过" body["reviewer_id"] = actor return _save(db, body, actor, action) if action == "release": if body.get("status") == "released": return body # 幂等：重复发布返回同一版本，不新建 blockers = ImprovementCandidate(**{**body, "released_ref": released_ref or {"id": "pending", "version": "0"}}).release_blockers() if blockers: raise NotValidated("；".join(blockers)) if publisher is not None and released_ref is None: released_ref = publisher(body) if not released_ref: raise NotValidated("发布必须引用一个既有的不可变版本，不得只改状态") body["status"] = "released" body["released_ref"] = released_ref return _save(db, body, actor, action) if action == "withdraw": if body.get("status") != "released": raise NotValidated("只有已发布的版本可以撤回；撤回只影响新运行") body["status"] = "withdrawn" return _save(db, body, actor, action) raise ValueError(f"未登记的改进动作：{action}") def listing(db, project_id: str, status: str | None = None) -> dict: rows = sorted(store.listing(db, KIND, project_id), key=lambda item: item.get("created_at") or "") if status: rows = [item for item in rows if item.get("status") == status] return {"items": rows}