qiaoqichao/mini-openclaw-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v2): T11 改进治理与演示交付——对照门禁、泄漏作废、三模块演示脚本

Browse Source 
覆盖 AC-15/16/34/37/41（BR-15/16/40、E-11/24）。

后端：
- contracts/improvement.py：ImprovementCandidate + ContrastReport/ContrastMetrics。
  零分母不得给比率或覆盖率（样本不足不是 0%）；非独立评测不得给改进/退化结论；
  已发布必须引用既有不可变版本；没有对照证据不得进入待审核或已发布。
- services/improvements.py：
  · evaluate_contrast 在服务端判定泄漏——预测锁定前读到标签即整次评测作废（E-11），
    保留集来源进开发集的样本作废并单列，指标按分子/可观察分母/覆盖率/未观察分开报；
  · transition 状态机：draft→evaluating→ready→in_review→approved→released，
    自批拒绝、退化与结论不明拒绝发布、发布幂等且只引用既有版本、撤回只影响新运行。
- api/prescreen.py：候选列表、提出与状态机三个端点；评测请求体提交样本，
  泄漏与指标一律由服务端计算，不接受客户端交来的"成品报告"。
- contracts/prescreen.py：补 E-11/E-24 错误码与 409 映射。

前端一致性修正：
- src/lib/prescreen-run.ts 的呈现判定改用后端既有词汇
  （empty/list/grouped/radar/faceted，与 services/review_plan.presentation 一致），
  避免前后端各一套判定导致界面画出后端判定为不该画雷达的图。

演示交付：
- scripts/xingbang_prescreen_demo.py：按 08 §7 八步演示，只驱动已实现入口，
  未实现的步骤（批量清单、历史提炼、卡片治理）如实标注原因；
  工程演示与模型效果分列，本地专项/Jev/真实概率一律标未验证或未接通。
- backend/test_demo_script.py：冒烟断言八步状态、未实现原因与分列口径。

验证：backend 全量 unittest 519 项 OK；前端 npm run build 通过。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
qiaoqichao
2026-09-30 09:25:25 +08:00
co-authored by Claude Code
parent f2bbac03e7
commit d6bc8485f5
10 changed files with 1018 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/api/prescreen.py
+79 -1
View File
@@ -33,6 +33,7 @@ from pydantic import BaseModel, ConfigDict, Field
from api.project_assistant import enabled
from api.risk_agent_access import authorize, check_project
from contracts.prescreen import (
DESC_MAX,
EVIDENCE_MAX,
ID_MAX,
MAX_BATCH_FILES,
@@ -42,7 +43,8 @@ from contracts.prescreen import (
S1_ERROR_RETRYABLE,
)
from services import assistant_store as store
from services import fact_conflict, material_import, prescreen_sandbox, prescreen_store, trajectory
from services import (fact_conflict, improvements, material_import, prescreen_sandbox,
prescreen_store, trajectory)
from services.assistant_settings import flags
from services.fact_conflict import FactConflictUnresolved
from services.project_assistant import ProjectAssistantService
@@ -92,6 +94,28 @@ class SimulationInput(Input):
expected_revision: int | None = Field(default=None, ge=1)
class ImprovementInput(Input):
"""T11 改进候选。必须说明改什么、依据哪些运行或结果。"""
target: str = Field(default="knowledge", max_length=40)
target_ref: dict
proposed_revision_ref: dict
change_summary: str = Field(min_length=1, max_length=DESC_MAX)
source_run_ids: list[str] = Field(default_factory=list, max_length=200)
outcome_ids: list[str] = Field(default_factory=list, max_length=200)
class TransitionInput(Input):
"""状态机动作。评测动作必须随请求提交对照结果，不能只改状态。"""
action: str = Field(min_length=1, max_length=40)
expected_revision: int | None = Field(default=None, ge=1)
evaluation: dict | None = None
contrast: dict | None = None
released_ref: dict | None = None
reason: str | None = Field(default=None, max_length=DESC_MAX)
class OutcomeInput(Input):
"""T10 会议结果。原始结果码必填；未提供时不得推断成通过或否决。"""
@@ -518,3 +542,57 @@ def record_review_outcome(project_id: str, payload: OutcomeInput, request: Reque
known_snapshots=snapshots)
return run(request, action)
# ---------------------------------------------------------------- 改进治理（T11）
@router.get("/projects/{project_id}/improvements")
def list_improvements(project_id: str, request: Request, status: str | None = None):
blocked = scope(request, project_id)
if blocked is not None:
return blocked
def action():
with _db() as db:
return improvements.listing(db, project_id, status)
return run(request, action)
@router.post("/projects/{project_id}/improvements", status_code=201)
def propose_improvement(project_id: str, payload: ImprovementInput, request: Request):
"""提出改进候选（草稿）。发布仍须经过对照、独立审核两个门。"""
blocked = scope(request, project_id)
if blocked is not None:
return blocked
def action():
with _db() as db:
project = _project(db, project_id)
return improvements.propose(db, project_id, payload.model_dump(),
_actor(request), project["data_domain"])
return run(request, action)
@router.post("/improvements/{candidate_id}/transitions")
def transition_improvement(candidate_id: str, payload: TransitionInput, request: Request):
"""候选状态机。泄漏按 E-11 拒绝，未通过对照/审核按 E-24 拒绝。"""
def action():
with _db() as db:
probe = store.get(db, improvements.KIND, candidate_id)
blocked = scope(request, probe["project_id"])
if blocked is not None:
return blocked
try:
with _db() as db:
return improvements.transition(
db, candidate_id, payload.action, _actor(request),
payload.expected_revision, evaluation=payload.evaluation,
contrast=payload.contrast, released_ref=payload.released_ref,
reason=payload.reason)
except improvements.LeakageDetected as exc:
return _envelope(request, ErrorCode.EVALUATION_CONTAMINATED, str(exc))
except improvements.NotValidated as exc:
return _envelope(request, ErrorCode.IMPROVEMENT_NOT_VALIDATED, str(exc))
return run(request, action)
backend/contracts/improvement.py
+147
View File
@@ -0,0 +1,147 @@
"""T11 改进候选契约：对照证据、独立审核与发布。
Owner：本文件（docs/v2/03-detailed-design.md §9.4 ImprovementCandidate 与状态机、
docs/v2/14-experience-loop.md §8、06-validation.md §5）。
三条不可让渡的规则：
1. **没有对照证据不能发布**：候选必须带一次冻结版本的旧/新对照，且不是退化
（AC-34 / 06 §5"未评测、自批、退化均不能发布"）。
2. **自批不算审核**：提交人与审核人必须是不同的人；同模型自评也不能当独立评测
（AC-41）。
3. **发布是引用不可变版本**：发布只把候选指向一个既有的、不可变的发布结果，
不允许"标成已发布却没有对应版本"（03 §9.4 末）。
"""
from __future__ import annotations
from typing import Literal
from pydantic import Field, model_validator
from contracts.prescreen import (
DESC_MAX,
ID_MAX,
AwareDatetime,
Contract,
DataDomain,
ErrorCode,
Ref,
)
CANDIDATE_STATUSES = ("draft", "evaluating", "ready", "in_review", "approved",
"rejected", "released", "withdrawn")
TARGETS = ("knowledge", "orchestration", "question_profile")
class ContrastMetrics(Contract):
"""高预测组指标口径（AC-37）：分子、可观察分母、覆盖率与未观察数分开报。
零分母时必须留空而不是填 0%——"没有样本"与"全都没通过"是两件事。
"""
label: str = Field(min_length=1, max_length=DESC_MAX)
numerator: int = Field(ge=0)
observable_denominator: int = Field(ge=0)
unobserved: int = Field(ge=0)
coverage: float | None = Field(default=None, ge=0, le=1)
rate: float | None = Field(default=None, ge=0, le=1)
@model_validator(mode="after")
def _check_metrics(self) -> ContrastMetrics:
if self.numerator > self.observable_denominator:
raise ValueError("分子不得大于可观察分母")
if self.observable_denominator == 0:
if self.rate is not None or self.coverage is not None:
raise ValueError("零分母时不得给出比率或覆盖率；样本不足不是 0%")
elif self.rate is None:
raise ValueError("有可观察样本时必须给出比率，避免用空白代替结果")
return self
class ContrastReport(Contract):
"""一次旧/新对照的结果。versions 冻结后才可比较（06 §5）。"""
id: str = Field(min_length=1, max_length=ID_MAX)
baseline_ref: Ref
candidate_ref: Ref
dataset_ref: str = Field(min_length=1, max_length=DESC_MAX)
split_ref: str = Field(min_length=1, max_length=DESC_MAX)
baseline_metrics: list[ContrastMetrics] = Field(default_factory=list, max_length=20)
candidate_metrics: list[ContrastMetrics] = Field(default_factory=list, max_length=20)
sample_size: int = Field(ge=0)
contaminated: list[str] = Field(default_factory=list, max_length=200)
independent: bool = False
verdict: Literal["improved", "unchanged", "regressed", "inconclusive"]
reviewer_id: str | None = Field(default=None, max_length=ID_MAX)
created_at: AwareDatetime
@model_validator(mode="after")
def _check_report(self) -> ContrastReport:
if self.verdict != "inconclusive" and not self.independent:
raise ValueError("非独立评测不得给出改进/退化结论，只能记 inconclusive")
if self.sample_size == 0 and self.verdict == "improved":
raise ValueError("没有样本不得声称改进")
return self
class ImprovementCandidate(Contract):
"""改进候选。发布引用既有不可变版本；撤回只影响新运行。"""
id: str = Field(min_length=1, max_length=ID_MAX)
project_id: str = Field(min_length=1, max_length=ID_MAX)
data_domain: DataDomain
revision: int = Field(ge=1)
target: Literal["knowledge", "orchestration", "question_profile"]
target_ref: Ref
source_run_ids: list[str] = Field(default_factory=list, max_length=200)
outcome_ids: list[str] = Field(default_factory=list, max_length=200)
change_summary: str = Field(min_length=1, max_length=DESC_MAX)
proposed_revision_ref: Ref
status: Literal["draft", "evaluating", "ready", "in_review", "approved",
"rejected", "released", "withdrawn"]
evaluation_ref: Ref | None = None
contrast: ContrastReport | None = None
reviewer_id: str | None = Field(default=None, max_length=ID_MAX)
released_ref: Ref | None = None
proposed_by: str = Field(min_length=1, max_length=ID_MAX)
rejection_reason: str | None = Field(default=None, max_length=DESC_MAX)
created_at: AwareDatetime
@model_validator(mode="after")
def _check_candidate(self) -> ImprovementCandidate:
if self.status in ("approved", "released") and self.reviewer_id is None:
raise ValueError("通过审核的候选必须记录审核人")
if self.status == "released" and self.released_ref is None:
raise ValueError("已发布必须引用一个既有的不可变发布版本，不得只改状态")
if self.status == "rejected" and not self.rejection_reason:
raise ValueError("拒绝必须留下原因")
if self.status in ("ready", "in_review", "approved", "released") \
and self.contrast is None:
raise ValueError("没有对照证据的候选不得进入待审核或已发布状态")
return self
def must_not_self_approve(self, actor: str) -> None:
if actor == self.proposed_by:
raise PermissionError("改进候选不能由提出者自己审核（自批不算独立审核）")
def release_blockers(self) -> list[str]:
"""发布前必须为空的阻塞项列表；可读的失败原因，便于界面展示。"""
blockers = []
if self.contrast is None:
blockers.append("缺少旧/新对照证据")
else:
if self.contrast.contaminated:
blockers.append(f"对照样本存在泄漏：{self.contrast.contaminated}")
if not self.contrast.independent:
blockers.append("对照不是独立评测（同模型自评不算证据）")
if self.contrast.verdict == "regressed":
blockers.append("对照显示退化，不得发布")
if self.contrast.verdict == "inconclusive":
blockers.append("对照结论不明确，不得发布")
if self.status != "approved":
blockers.append(f"当前状态为 {self.status}，未通过审核")
return blockers
CONTAMINATION_CODE = ErrorCode.EVALUATION_CONTAMINATED
NOT_VALIDATED_CODE = ErrorCode.IMPROVEMENT_NOT_VALIDATED
backend/contracts/prescreen.py
+5
View File
@@ -109,6 +109,9 @@ class ErrorCode(StrEnum):
DIMENSION_SCHEMA_CONFLICT = "DIMENSION_SCHEMA_CONFLICT"
DEMO_ESTIMATE_FORBIDDEN = "DEMO_ESTIMATE_FORBIDDEN"
SCENARIO_ASSUMPTION_CONFLICT = "SCENARIO_ASSUMPTION_CONFLICT"
# T11 改进治理：泄漏与未验证都是"先改数据再重试"的冲突，走 409。
EVALUATION_CONTAMINATED = "EVALUATION_CONTAMINATED"
IMPROVEMENT_NOT_VALIDATED = "IMPROVEMENT_NOT_VALIDATED"
# 运行级编码：记录在运行体（degradations / judgment）里，**不构成 HTTP 错误**，
# 因此刻意不进 S1_ERROR_HTTP；见下方 RUN_LEVEL_CODES。
JUDGMENT_UNAVAILABLE = "JUDGMENT_UNAVAILABLE"
@@ -140,6 +143,8 @@ S1_ERROR_HTTP: dict[ErrorCode, int] = {
ErrorCode.DIMENSION_SCHEMA_CONFLICT: 422,
ErrorCode.DEMO_ESTIMATE_FORBIDDEN: 422,
ErrorCode.SCENARIO_ASSUMPTION_CONFLICT: 422,
ErrorCode.EVALUATION_CONTAMINATED: 409,
ErrorCode.IMPROVEMENT_NOT_VALIDATED: 409,
}
# 这些编码只出现在运行结果里（"HTTP 200 仅表示结果可读取"），永远不会作为
backend/services/improvements.py
+204
View File
@@ -0,0 +1,204 @@
"""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}
backend/test_demo_script.py
+58
View File
@@ -0,0 +1,58 @@
"""T11 演示脚本冒烟测试：八步都出现、未实现处不伪装、工程与模型效果分列。"""
import contextlib
import io
import sys
import unittest
from pathlib import Path
SCRIPTS = Path(__file__).resolve().parent.parent / "scripts"
sys.path.insert(0, str(SCRIPTS))
import xingbang_prescreen_demo as demo # noqa: E402
class DemoScriptTests(unittest.TestCase):
def output(self) -> str:
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer):
code = demo.main()
self.assertEqual(code, 0)
return buffer.getvalue()
def test_every_step_is_reported_with_its_real_status(self):
text = self.output()
for index, title, status, _ in demo.STEPS:
self.assertIn(f"{index}. [{status}] {title}", text)
def test_unimplemented_steps_say_why(self):
text = self.output()
self.assertIn("未实现", text)
self.assertIn("需要真实历史材料", text)
def test_engineering_demo_and_model_effects_are_separate_sections(self):
text = self.output()
self.assertIn("工程演示", text)
self.assertIn("模型效果", text)
self.assertIn("未验证（无权重与授权）", text)
self.assertIn("Jev：未接通", text)
self.assertIn("不能作为真实业务效果或审批依据", text)
def test_real_probability_stays_empty_and_demo_values_are_labeled(self):
facts = demo.run_demo()
self.assertTrue(facts["probability_is_null"], "真实概率必须为空")
self.assertEqual(facts["sandbox"]["baseline"], 0.42)
self.assertEqual(facts["sandbox"]["combined"], 0.64)
self.assertIn("人工设定", facts["sandbox"]["label"])
self.assertEqual(facts["sandbox"]["unchanged_dimensions"], ["DC-DEAL"])
def test_trajectory_keeps_failures_and_excludes_unmatched_outcomes(self):
facts = demo.run_demo()
self.assertTrue(facts["trajectory"]["failed_preserved"])
self.assertEqual(facts["trajectory"]["unmatched"], 1)
self.assertEqual(facts["trajectory"]["observed_denominator"], 0,
"未匹配材料版本的结果不得进入已观察分母")
if __name__ == "__main__":
unittest.main()
backend/test_improvements.py
+286
View File
@@ -0,0 +1,286 @@
"""T11 改进治理测试：对照、泄漏、独立审核与发布门禁。
覆盖 AC-15/16/34/37/41（BR-15/16/40、E-11/24）。
"""
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.testclient import TestClient
from api import prescreen as api
from contracts.improvement import ContrastMetrics, ImprovementCandidate
from services import assistant_store as store
from services import improvements
from services.project_assistant import ProjectAssistantService
HEADERS = {"X-Xingbang-Demo-Role": "manager"}
PROJECT = "DEMO-B"
BASELINE = {"id": "assistant.current", "version": "1.0.0"}
CANDIDATE = {"id": "prescreen.v2", "version": "0.2.0"}
def sample(sample_id, *, split="dev", observed=True, baseline_pass=True,
candidate_pass=True, predicted="2026-09-29T01:00:00+00:00",
revealed="2026-09-29T03:00:00+00:00", source_in_dev=False):
return {"id": sample_id, "split": split, "observed": observed,
"baseline_pass": baseline_pass, "candidate_pass": candidate_pass,
"predicted_at": predicted, "label_revealed_at": revealed,
"source_in_dev": source_in_dev}
def contrast(samples, *, independent=True, reviewer="reviewer-1"):
return improvements.evaluate_contrast(
samples, BASELINE, CANDIDATE, dataset_ref="fixtures/cases/manifest.json",
split_ref="split.v1", reviewer_id=reviewer, independent=independent)
class ContrastTests(unittest.TestCase):
def test_labels_read_before_the_lock_void_the_evaluation(self):
with self.assertRaises(improvements.LeakageDetected):
contrast([sample("S-1", predicted="2026-09-29T05:00:00+00:00")])
def test_holdout_source_used_in_dev_is_contaminated(self):
report = contrast([sample("S-1"), sample("S-2", split="holdout", source_in_dev=True)])
self.assertEqual(report["contaminated"], ["S-2"])
self.assertEqual(report["sample_size"], 1, "泄漏样本作废，不参与指标")
def test_metrics_keep_numerator_denominator_and_unobserved_apart(self):
report = contrast([sample("S-1"), sample("S-2", observed=False),
sample("S-3", baseline_pass=False, candidate_pass=True)])
baseline = report["baseline_metrics"][0]
candidate = report["candidate_metrics"][0]
self.assertEqual((baseline["numerator"], baseline["observable_denominator"],
baseline["unobserved"]), (1, 2, 1))
self.assertEqual((candidate["numerator"], candidate["observable_denominator"]), (2, 2))
self.assertEqual(baseline["coverage"], 0.6667)
self.assertEqual(report["verdict"], "improved")
def test_zero_denominator_is_not_reported_as_zero_percent(self):
report = contrast([sample("S-1", observed=False)])
self.assertIsNone(report["candidate_metrics"][0]["rate"])
self.assertIsNone(report["candidate_metrics"][0]["coverage"])
self.assertEqual(report["verdict"], "inconclusive")
def test_non_independent_evaluation_cannot_claim_improvement(self):
report = contrast([sample("S-1")], independent=False)
self.assertEqual(report["verdict"], "inconclusive")
def test_contract_refuses_rate_without_a_denominator(self):
with self.assertRaises(ValueError):
ContrastMetrics(label="高预测组", numerator=0, observable_denominator=0,
unobserved=3, rate=0.0)
def test_metrics_cannot_have_more_passes_than_samples(self):
with self.assertRaises(ValueError):
ContrastMetrics(label="高预测组", numerator=3, observable_denominator=2,
unobserved=0, rate=1.0)
class ImprovementLifecycleTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
self.service = ProjectAssistantService(Path(self.temp.name) / "assistant.sqlite3")
self.published: list[dict] = []
def tearDown(self):
self.temp.cleanup()
def publish(self, body):
ref = {"id": f"card-{len(self.published) + 1}", "version": "2.0.0"}
self.published.append(ref)
return ref
def propose(self, actor="manager-1"):
with store.transaction(self.service.db_path) as db:
return improvements.propose(db, PROJECT, {
"target": "knowledge", "target_ref": {"id": "card-1", "version": "1.0.0"},
"proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
"change_summary": "把回款核验的经验扩到制造业设备直租",
"source_run_ids": ["run-1"]}, actor, "demo")
def transition(self, candidate, action, actor="reviewer-9", **over):
with store.transaction(self.service.db_path) as db:
return improvements.transition(db, candidate["id"], action, actor,
over.pop("expected_revision", None), **over)
def test_proposal_without_sources_is_rejected(self):
with self.assertRaises(ValueError):
with store.transaction(self.service.db_path) as db:
improvements.propose(db, PROJECT, {
"target": "knowledge", "target_ref": {"id": "c", "version": "1"},
"proposed_revision_ref": {"id": "c", "version": "2"},
"change_summary": "改一下"}, "manager-1", "demo")
def test_release_requires_contrast_evidence(self):
candidate = self.propose()
with self.assertRaises(improvements.NotValidated) as ctx:
self.transition(candidate, "release", released_ref={"id": "x", "version": "1"})
self.assertIn("未通过审核", str(ctx.exception))
def test_improved_candidate_can_be_approved_and_released(self):
candidate = self.propose()
report = contrast([sample("S-1", candidate_pass=True),
sample("S-2", baseline_pass=False, candidate_pass=True)])
evaluated = self.transition(candidate, "evaluate", contrast=report)
self.assertEqual(evaluated["status"], "ready")
self.transition(evaluated, "submit")
approved = self.transition(evaluated, "approve", actor="reviewer-9")
self.assertEqual(approved["status"], "approved")
released = self.transition(approved, "release", publisher=self.publish)
self.assertEqual(released["status"], "released")
self.assertEqual(released["released_ref"], {"id": "card-1", "version": "2.0.0"})
self.assertEqual(len(self.published), 1)
def test_self_approval_is_refused(self):
candidate = self.propose(actor="manager-1")
report = contrast([sample("S-1")])
self.transition(candidate, "evaluate", contrast=report)
self.transition(candidate, "submit")
with self.assertRaises(PermissionError):
self.transition(candidate, "approve", actor="manager-1")
def test_regressed_candidate_cannot_be_released(self):
candidate = self.propose()
report = contrast([sample("S-1", baseline_pass=True, candidate_pass=False)])
self.assertEqual(report["verdict"], "regressed")
evaluated = self.transition(candidate, "evaluate", contrast=report)
self.assertEqual(evaluated["status"], "evaluating")
with self.assertRaises(improvements.NotValidated):
self.transition(evaluated, "release", released_ref={"id": "x", "version": "1"})
def test_contaminated_evaluation_does_not_advance_the_candidate(self):
candidate = self.propose()
report = contrast([sample("S-1"), sample("S-2", split="holdout", source_in_dev=True)])
evaluated = self.transition(candidate, "evaluate", contrast=report)
self.assertEqual(evaluated["status"], "evaluating")
self.assertIn("泄漏", evaluated["rejection_reason"])
def test_release_is_idempotent(self):
candidate = self.propose()
report = contrast([sample("S-1")])
evaluated = self.transition(candidate, "evaluate", contrast=report)
self.transition(evaluated, "submit")
approved = self.transition(evaluated, "approve")
released = self.transition(approved, "release", publisher=self.publish)
again = self.transition(released, "release", publisher=self.publish)
self.assertEqual(again, released, "重复发布不得新建版本")
self.assertEqual(len(self.published), 1)
def test_withdraw_only_applies_to_released_versions(self):
candidate = self.propose()
with self.assertRaises(improvements.NotValidated):
self.transition(candidate, "withdraw")
def test_contract_refuses_released_without_a_version_reference(self):
body = {"id": "imp-1", "project_id": PROJECT, "data_domain": "demo", "revision": 1,
"target": "knowledge", "target_ref": {"id": "c", "version": "1"},
"change_summary": "改", "proposed_revision_ref": {"id": "c", "version": "2"},
"status": "released", "reviewer_id": "r", "proposed_by": "m",
"created_at": "2026-09-30T01:00:00+00:00"}
with self.assertRaises(ValueError):
ImprovementCandidate(**body)
class ImprovementApiTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
self.service = ProjectAssistantService(Path(self.temp.name) / "assistant.sqlite3")
self.patches = [
patch.object(api, "_service", lambda: self.service),
patch.dict(os.environ, {"XINGBANG_AUTH_USERS": "", "XINGBANG_REQUIRE_AUTH": "0",
"XINGBANG_LOCAL_MODEL_URL": ""}),
]
for item in self.patches:
item.start()
with store.transaction(self.service.db_path) as db:
store.put(db, "project", {"id": PROJECT, "data_domain": "demo", "revision": 1})
app = FastAPI()
app.add_exception_handler(RequestValidationError, api.validation_handler)
app.include_router(api.router)
self.client = TestClient(app)
self.base = "/api/xingbang/risk-agent/project-assistant/prescreen"
def tearDown(self):
self.client.close()
for item in reversed(self.patches):
item.stop()
self.temp.cleanup()
def test_propose_then_blocked_release(self):
created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
json={"target_ref": {"id": "card-1", "version": "1.0.0"},
"proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
"change_summary": "扩适用范围",
"source_run_ids": ["run-1"]}, headers=HEADERS)
self.assertEqual(created.status_code, 201, created.text)
candidate = created.json()
self.assertEqual(candidate["status"], "draft")
blocked = self.client.post(
f"{self.base}/improvements/{candidate['id']}/transitions",
json={"action": "release",
"released_ref": {"id": "card-1", "version": "2.0.0"}}, headers=HEADERS)
self.assertEqual(blocked.status_code, 409)
self.assertEqual(blocked.json()["error"]["code"], "IMPROVEMENT_NOT_VALIDATED")
def test_leakage_is_reported_as_e11(self):
created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
json={"target_ref": {"id": "card-1", "version": "1.0.0"},
"proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
"change_summary": "扩适用范围",
"outcome_ids": ["outcome-1"]}, headers=HEADERS).json()
# 泄漏在服务端判定：预测锁定前已读到标签的样本使整次评测作废
leaked = self.client.post(
f"{self.base}/improvements/{created['id']}/transitions",
json={"action": "evaluate", "evaluation": {
"baseline_ref": BASELINE, "candidate_ref": CANDIDATE,
"dataset_ref": "fixtures/cases/manifest.json", "split_ref": "split.v1",
"independent": True,
"samples": [sample("S-1", predicted="2026-09-29T05:00:00+00:00")]}},
headers=HEADERS)
self.assertEqual(leaked.status_code, 409, leaked.text)
self.assertEqual(leaked.json()["error"]["code"], "EVALUATION_CONTAMINATED")
clean = self.client.post(
f"{self.base}/improvements/{created['id']}/transitions",
json={"action": "evaluate", "evaluation": {
"baseline_ref": BASELINE, "candidate_ref": CANDIDATE,
"dataset_ref": "fixtures/cases/manifest.json", "split_ref": "split.v1",
"independent": True, "samples": [sample("S-1")]}},
headers=HEADERS)
self.assertEqual(clean.status_code, 200, clean.text)
self.assertEqual(clean.json()["status"], "ready")
def test_evaluation_without_samples_is_rejected(self):
created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
json={"target_ref": {"id": "card-1", "version": "1.0.0"},
"proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
"change_summary": "扩适用范围",
"source_run_ids": ["run-1"]}, headers=HEADERS).json()
response = self.client.post(
f"{self.base}/improvements/{created['id']}/transitions",
json={"action": "evaluate", "contrast": {}}, headers=HEADERS)
self.assertEqual(response.status_code, 422, response.text)
self.assertEqual(response.json()["error"]["code"], "INVALID_REQUEST")
def test_unknown_action_is_rejected(self):
created = self.client.post(f"{self.base}/projects/{PROJECT}/improvements",
json={"target_ref": {"id": "card-1", "version": "1.0.0"},
"proposed_revision_ref": {"id": "card-1", "version": "2.0.0"},
"change_summary": "扩适用范围",
"source_run_ids": ["run-1"]}, headers=HEADERS).json()
response = self.client.post(
f"{self.base}/improvements/{created['id']}/transitions",
json={"action": "promote"}, headers=HEADERS)
self.assertEqual(response.status_code, 422)
self.assertEqual(response.json()["error"]["code"], "INVALID_REQUEST")
if __name__ == "__main__":
unittest.main()
scripts/test_prescreen_ui.cjs
+4 -3
View File
@@ -43,10 +43,11 @@ assert.match(lib, /domain !== "demo"/, "lib 必须在非演示域返回 null");
// 5. 雷达只对同刻度 3–8 维绘制，其余给完整列表；空维度不补占位轴（AC-44）
assert.match(lib, /RADAR_MIN = 3/);
assert.match(lib, /RADAR_MAX = 8/);
assert.match(lib, /kind: "radar"/);
assert.match(lib, /layout: "radar"/);
assert.match(lib, /DimensionLayout = "empty" \| "list" \| "grouped" \| "radar" \| "faceted"/, "呈现词汇必须与后端 review_plan.presentation 一致");
assert.match(lib, /EMPTY_DIMENSIONS = "当前方案尚未配置核查维度"/);
assert.match(panel, /shown\.kind === "radar" && <RadarChart/, "只有判定为雷达时才画雷达");
assert.match(panel, /shown\.kind === "empty" && <div className="ra-empty">/, "空维度用空状态，不补六轴");
assert.match(panel, /shown\.layout === "radar" && <RadarChart/, "只有判定为雷达时才画雷达");
assert.match(panel, /shown\.layout === "empty" && <div className="ra-empty">/, "空维度用空状态，不补六轴");
assert.match(panel, /文字列表与图等价/, "图必须有等价文字列表");
// 6. 乱序与过期响应必须丢弃（AC-31）
scripts/xingbang_prescreen_demo.py
+218
View File
@@ -0,0 +1,218 @@
"""兴邦预审 P0 模拟闭环演示脚本（T11）。
依据：docs/v2/08-data-pilot-and-demo.md §7 的八步演示顺序。
**诚实性要求**（08 §7 末、06 §6）：脚本只驱动本仓库已经实现并验证过的入口；
没有实现的步骤打印"未实现"及其原因，不伪造屏幕状态；模型效果与工程演示
分列，模拟标签不得包装成真实业务验证。
用法：backend/venv/Scripts/python.exe scripts/xingbang_prescreen_demo.py
全程在临时目录内使用临时数据库，不触碰真实 xingbang_data。
"""
from __future__ import annotations
import sys
import tempfile
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "backend"))
from contracts.prescreen import ErrorCode # noqa: E402
from services import (assistant_store as store, improvements, judgment_adapters, # noqa: E402
judgment_provider as jp, prescreen, prescreen_policy,
prescreen_sandbox, prescreen_store, review_plan, trajectory)
DONE = "已执行"
NOT_DONE = "未实现"
MODEL_UNVERIFIED = "未验证（无权重与授权）"
FINDINGS = [
{"code": "corporate.cashflow", "result": "unknown", "stage": 1, "evidence_ids": ["EV-1"]},
{"code": "corporate.title", "result": "unknown", "stage": 1, "evidence_ids": ["EV-2"]},
{"code": "corporate.deal", "result": "checked", "stage": 2, "evidence_ids": ["EV-3"]},
]
DIMENSIONS = [
{"id": "DC-CASH", "component_refs": [{"id": "corporate.cashflow"}]},
{"id": "DC-TITLE", "component_refs": [{"id": "corporate.title"}]},
{"id": "DC-DEAL", "component_refs": [{"id": "corporate.deal"}]},
]
PLAN = {
"plan": {"id": "corporate.default", "version": "1.0.0"},
"skills": [{"id": "corporate.cashflow", "version": "1.0.0"}],
"rules": [], "template": {"id": "template.ai-pre-review", "version": "1.0.0"},
"content_hash": "demo-hash", "dimensions": DIMENSIONS,
}
STEPS = [
(1, "3,000 条候选清单、生成版本与多结果分布", NOT_DONE,
"T04 的服务已在单测中验证，但本脚本不重复驱动批量生成；清单需真实数据授权"),
(2, "选开发集案例，展开问询/补件/会议事件，提炼 Skill 与 advisory Rule", NOT_DONE,
"历史提炼走 T05 的治理入口，需要真实历史材料；此处不伪造提炼结果"),
(3, "自然语言改适用范围 + 人工新增卡片，统一审核发布", NOT_DONE,
"卡片治理入口属 T05；演示脚本不代替审核动作"),
(4, "数据卡 + Skills + Rules 组方案，维度随方案变化，Tools 只在依据中展示", DONE, None),
(5, "缺证 → 主审核查 → 问题与补件 → 演示概率来源", DONE, None),
(6, "勾选动作比较情景、撤销恢复基线", DONE, None),
(7, "回放完整 trajectory，对齐模拟会议记录，查看材料版本差异", DONE, None),
(8, "从漏检提出改进候选，隔离样本对照后人工审核发布，旧版仍可回放", DONE, None),
]
def run_demo() -> dict:
facts: dict = {"steps": [], "sandbox": {}, "trajectory": {}, "improvement": {}}
with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmp:
db_path = Path(tmp) / "demo.sqlite3"
store.initialize(db_path)
# 步骤 4：方案维度（去重与刻度判定）
dimensions = review_plan.resolve_dimensions([
{"id": "DC-CASH", "label": "回款核验", "scale_id": "evidence-stage",
"scale_version": "1.0.0", "order": 1, "component_refs": [{"id": "corporate.cashflow", "version": "1.0.0"}],
"levels": [{"value": i, "label": name, "criterion": f"证据阶段：{name}"} for i, name in
enumerate(("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核"))]},
{"id": "DC-TITLE", "label": "权属核验", "scale_id": "evidence-stage",
"scale_version": "1.0.0", "order": 2, "component_refs": [{"id": "corporate.title", "version": "1.0.0"}],
"levels": [{"value": i, "label": name, "criterion": f"证据阶段：{name}"} for i, name in
enumerate(("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核"))]},
{"id": "DC-DEAL", "label": "交易要素", "scale_id": "evidence-stage",
"scale_version": "1.0.0", "order": 3, "component_refs": [{"id": "corporate.deal", "version": "1.0.0"}],
"levels": [{"value": i, "label": name, "criterion": f"证据阶段：{name}"} for i, name in
enumerate(("未提供", "仅有说明", "资料待核", "交叉核验", "人工复核"))]},
])
presentation = review_plan.presentation(dimensions)
facts["presentation"] = presentation["layout"]
# 步骤 5：预审执行（专项关闭，无校准概率）
config = jp.resolve_config(False)
decision = prescreen_policy.decide(FINDINGS)
decision["findings"] = FINDINGS
decision["missing_inputs"] = ["近六个月银行流水", "设备权属材料"]
screening = prescreen.assemble(
{"id": "run-demo", "project_id": "DEMO-026", "revision": 1,
"created_at": store.now(), "input_snapshot_id": "snap-demo"},
PLAN, decision, config)
with store.transaction(db_path) as db:
prescreen_store.save(db, screening)
facts["recommendation"] = screening["recommendation"]
facts["probability_is_null"] = screening["probability"] is None
# 步骤 6：沙盘（演示值只在演示域、组合查表）
with store.transaction(db_path) as db:
stored = prescreen_store.load(db, "DEMO-026", "run-demo")
baseline = prescreen_sandbox.simulate(stored, [])
both = prescreen_sandbox.simulate(stored, ["assume_bank_flow", "assume_title"])
facts["sandbox"] = {
"baseline": baseline["probability_view"]["demo_estimate"]["value"],
"combined": both["probability_view"]["demo_estimate"]["value"],
"label": baseline["probability_view"]["demo_estimate"]["label"],
"unchanged_dimensions": [item["id"] for item in both["dimensions"]
if not item["assumed"]],
}
# 步骤 7：轨迹回放与模拟会议结果
with store.transaction(db_path) as db:
store.put(db, "run", {"id": "run-demo", "project_id": "DEMO-026",
"status": "completed", "input_snapshot_id": "snap-demo",
"created_at": store.now(), "steps": [
{"run_id": "run-demo", "step_id": "corporate.cashflow",
"skill": {"id": "corporate.cashflow", "version": "1.0.0"},
"status": "completed", "input_snapshot_id": "snap-demo",
"started_at": store.now(), "outputs": {"note": "已核查"},
"findings": [{"code": "cashflow", "result": "unknown",
"evidence_ids": ["EV-1"]}],
"errors": [], "missing_inputs": []},
{"run_id": "run-demo", "step_id": "corporate.title",
"skill": {"id": "corporate.title", "version": "1.0.0"},
"status": "failed", "input_snapshot_id": "snap-demo",
"started_at": store.now(), "outputs": {},
"findings": [], "missing_inputs": [],
"errors": [{"code": "SKILL_TIMEOUT",
"message": "演示中的失败步骤"}]}]})
replay = trajectory.replay(db, "DEMO-026", "run-demo")
trajectory.record(db, "DEMO-026",
{"raw_code": "附条件通过", "meeting_id": "M-DEMO",
"occurred_at": store.now()}, "demo-reviewer",
known_snapshots=["snap-demo"])
outcomes = trajectory.listing(db, "DEMO-026")
facts["trajectory"] = {
"steps": [item["status"] for item in replay["steps"]],
"failed_preserved": any(item["status"] == "failed" for item in replay["steps"]),
"unmatched": len(outcomes["unmatched"]),
"observed_denominator": outcomes["counts_toward_denominator"]["observed"],
}
# 步骤 8：改进候选（合成样本、独立评测、可回放旧版）
with store.transaction(db_path) as db:
candidate = improvements.propose(db, "DEMO-026", {
"target": "knowledge", "target_ref": {"id": "card-demo", "version": "1.0.0"},
"proposed_revision_ref": {"id": "card-demo", "version": "1.1.0"},
"change_summary": "把回款核验经验扩到设备直租",
"source_run_ids": ["run-demo"]}, "demo-manager", "demo")
evaluation = {"baseline_ref": {"id": "current", "version": "1.0.0"},
"candidate_ref": {"id": "prescreen.v2", "version": "0.2.0"},
"dataset_ref": "demo/synthetic-manifest.json",
"split_ref": "demo-split.v1", "independent": True,
"samples": [
{"id": "S-1", "split": "dev", "observed": True,
"baseline_pass": True, "candidate_pass": True,
"predicted_at": "2026-09-29T01:00:00+00:00",
"label_revealed_at": "2026-09-29T03:00:00+00:00"},
{"id": "S-2", "split": "dev", "observed": True,
"baseline_pass": False, "candidate_pass": True,
"predicted_at": "2026-09-29T01:00:00+00:00",
"label_revealed_at": "2026-09-29T03:00:00+00:00"}]}
evaluated = improvements.transition(db, candidate["id"], "evaluate", "demo-reviewer",
evaluation=evaluation)
improvements.transition(db, evaluated["id"], "submit", "demo-reviewer")
approved = improvements.transition(db, evaluated["id"], "approve", "demo-reviewer")
released = improvements.transition(
db, approved["id"], "release", "demo-reviewer",
publisher=lambda body: {"id": "card-demo", "version": "1.1.0"})
facts["improvement"] = {"verdict": evaluated["contrast"]["verdict"],
"status": released["status"],
"released_ref": released["released_ref"]}
# 专项：离线 fixture 可跑，live 未接通
facts["judgment_offline"] = judgment_adapters.load_fixture(
"question_bank_extraction.json")["rubric_version"]
facts["demo_table_version"] = prescreen_sandbox.DEMO_TABLE_VERSION
facts["budget_code"] = ErrorCode.JUDGMENT_CONTEXT_EXCEEDED.value
return facts
def main() -> int:
facts = run_demo()
print("兴邦预审 P0 模拟闭环演示（合成数据 · 不连接真实数据与模型）")
print("=" * 78)
for index, title, status, reason in STEPS:
print(f"{index}. [{status}] {title}")
if reason:
print(f" 原因：{reason}")
print("-" * 78)
print("工程演示（本仓库已实现并验证的入口）")
print(f" · 维度呈现：{facts['presentation']}")
print(f" · 主审意见：{facts['recommendation']} · 真实概率为空：{facts['probability_is_null']}")
sandbox = facts["sandbox"]
print(f" · 沙盘：基线 {sandbox['baseline']:.0%} → 组合 {sandbox['combined']:.0%}"
f"（{sandbox['label']}）")
print(f" · 未变化的维度仍列出：{sandbox['unchanged_dimensions']}")
trace = facts["trajectory"]
print(f" · 轨迹步骤状态：{trace['steps']} · 失败步骤保留：{trace['failed_preserved']}")
print(f" · 会议结果：unmatched {trace['unmatched']} 条 · 已观察分母 {trace['observed_denominator']}")
imp = facts["improvement"]
print(f" · 改进对照：{imp['verdict']} → {imp['status']}（{imp['released_ref']}）")
print("-" * 78)
print("模型效果（与上面的工程演示分开计）")
print(f" · 本地专项：{MODEL_UNVERIFIED}；当前只有离线 fixture（{facts['judgment_offline']}）")
print(f" · Jev：未接通（无端点与凭据）")
print(f" · 真实审批概率：未校准，恒为空；界面上的百分数来自演示表 "
f"{facts['demo_table_version']}，不是模型输出")
print(f" · 超预算时的拒绝码（{facts['budget_code']}）证明预算守卫生效，与概率无关")
print("=" * 78)
print("以上均为合成数据与演示值，不能作为真实业务效果或审批依据。")
return 0
if __name__ == "__main__":
raise SystemExit(main())
src/components/xingbang/PrescreenPanel.tsx
+3 -3
View File
@@ -159,11 +159,11 @@ export default function PrescreenPanel({project, request, busy, setBusy, setErro
</div>
<h3>核查维度（{dimensions.length}）</h3>
{shown.kind === "empty" && <div className="ra-empty"><h2>{EMPTY_DIMENSIONS}</h2>
{shown.layout === "empty" && <div className="ra-empty"><h2>{EMPTY_DIMENSIONS}</h2>
<p>方案没有维度时不补占位轴，也不给出任何结论。</p></div>}
{shown.kind !== "empty" && <>
{shown.layout !== "empty" && <>
<p>{shown.reason}</p>
{shown.kind === "radar" && <RadarChart dimensions={dimensions} scenarioStages={scenarioStages} />}
{shown.layout === "radar" && <RadarChart dimensions={dimensions} scenarioStages={scenarioStages} />}
<div className="xb-finding">
<div>
{dimensions.map((item) => {
src/lib/prescreen-run.ts
+14 -7
View File
@@ -49,15 +49,22 @@ export const RUN_STATUS_LABELS: Record<string, string> = {
completed: "已完成", partial: "部分完成", failed: "未完成", cancelled: "已取消",
};
/** 维度呈现方式。同刻度 3–8 维才画雷达；其余给完整列表，绝不补占位轴（AC-44）。 */
export function presentationFor(dimensions: {scale_id?: string}[]): {kind: "empty" | "radar" | "bars" | "list" | "faceted"; reason: string} {
/** 维度呈现方式。同刻度 3–8 维才画雷达；其余给完整列表，绝不补占位轴（AC-44）。
*
* 词汇与后端 services/review_plan.presentation() 保持一致
* （empty/list/grouped/radar/faceted），避免前后端各有一套判定导致界面
* 画出后端判定为"不该画雷达"的图。
*/
export type DimensionLayout = "empty" | "list" | "grouped" | "radar" | "faceted";
export function presentationFor(dimensions: {scale_id?: string}[]): {layout: DimensionLayout; reason: string} {
const total = dimensions.length;
if (!total) return {kind: "empty", reason: EMPTY_DIMENSIONS};
if (!total) return {layout: "empty", reason: EMPTY_DIMENSIONS};
const scales = new Set(dimensions.map((item) => item.scale_id || ""));
if (scales.size > 1) return {kind: "faceted", reason: "维度刻度不一致，按分面展示；不强行归一化，避免比较不可比的量"};
if (total < RADAR_MIN) return {kind: "bars", reason: `仅 ${total} 个维度，用状态列表比雷达更清楚`};
if (total > RADAR_MAX) return {kind: "list", reason: `${total} 个维度超出雷达可读范围，改用分组列表且全部维度仍可查看`};
return {kind: "radar", reason: "同一刻度且 3–8 维，可用雷达并附等价文字列表"};
if (scales.size > 1) return {layout: "faceted", reason: "维度刻度不一致，按分面展示；不强行归一化，避免比较不可比的量"};
if (total < RADAR_MIN) return {layout: "list", reason: `仅 ${total} 个维度，用状态列表比雷达更清楚`};
if (total > RADAR_MAX) return {layout: "grouped", reason: `${total} 个维度超出雷达可读范围，改用分组列表且全部维度仍可查看`};
return {layout: "radar", reason: "同一刻度且 3–8 维，可用雷达并附等价文字列表"};
}
export function stageName(stage: number | null | undefined): string {