qiaoqichaoandClaude Code
2e7e19471c
覆盖 AC-50/51/52（BR-48/49、E-28）。审计确认的 P1-3 缺口"历史提炼与卡片 不在实时 API 调用图上"由此关闭：提炼、候选治理、卡片进方案全部有真实调用方。 后端： - contracts：新增 E-28 EXTRACTION_UNAVAILABLE（运行级，不进 HTTP 信封）与 KnowledgeExtraction 记录类型。契约层强制"失败不得带候选"， AC-50「模型不可用⇒不产生候选」是类型约束而不是文档约定。 - services/experience_extraction.py（新）：真实提炼入口。模型不可用或调用失败 → 记 E-28 且 candidate_id 为空；不产半成品候选。传输纪律同 assistant_model （禁进程内代理、零重试），且 tools=[] 使"越权读材料"在结构上不可能。 自然语言修订同样必须真实调用模型，没有"把中文猜成条件"的回退路径。 - services/knowledge_cards.py：新增 current_draft/revise/list_cards 与 registry_conditions。修订只产草稿、保留 origin 与来源、升版本记 parent_version； test_cases 纳入可改字段——它是 component_evaluation 对自定义 Skill 的发布门禁， 不开放则提炼出的候选永远卡在评测门外。 - services/project_assistant.py：enqueue 改用 review_plan.assemble()， 让 T06-b 的依赖预检（E-19 可定位文案）与维度解析不再被绕过。 - api/prescreen.py：case-timeline、knowledge-extractions、knowledge-candidates （创建/修订/状态机）、review-plans（装配/状态机）。角色门禁与既有 /components 入口一致；id 与数据域由服务端生成；按 id 的路由一律在 run() 内 查归属，否则 KeyError 会变成 500 而不是 404。 - component_registry/review_plan：下沉 current_draft/version_key 与 list_plans。 前端： - 新增 /experience-center 独立入口（BR-50）与 ExperienceCenter 组件； AgentNav 增加「历史经验加工」。 - originBadge 此前零调用方、.origin-* 无任何 CSS——守卫只 grep 库文件字符串， 于是"测试通过但功能不存在"。本批补齐消费方与样式，并用左边框样式区分四种来源， 保证去色仍可区分（AC-42/BR-48）。 - 守卫语义按 AC-55 调整：由"禁止新建路由"改为"禁止重复页面" （面板不得搬到独立路由、已合并页必须保持重定向），并新增 experience-center-ui 守卫（已登记 verify.py，共 12 项检查）。 测试：新增 36 项（提炼 15 / 候选治理 15 / 卡片到运行 6），全量 unittest 无回归； 两次同项目运行对照证明卡片确实改变了核查结果，且 basis_ids 指向 卡片id@版本。 未闭合（记录在 17 的剩余缺口）：真实模型提炼尚未实测（需端点）； 界面未经浏览器点击验证；例外边界未进入可求值语言；规则类候选不在本批。 Co-Authored-By: Claude Code <noreply@anthropic.com>
290 lines
14 KiB
Python
290 lines
14 KiB
Python
"""纠偏第二批 B-b · 候选治理与自然语言修订测试（AC-51 / BR-42 / AC-43）。
|
|
|
|
三条必须机械成立的约束：
|
|
|
|
1. **自然语言修订必须真的调用模型**——没有"把中文猜成条件"的确定性回退；
|
|
模型不可用时拒绝且**卡片一字未改**（反例断言比对的是修订前后的卡片内容）。
|
|
2. **只产草稿**：修订后状态仍是 draft，绝不因"改过"而生效。
|
|
3. **来源不可洗 + 不自批**：历史卡片修订后仍是 history_derived 且保留来源事件；
|
|
发布需要独立审核人且先过隔离评测。
|
|
|
|
用临时库；不触碰真实数据根。模型用假传输层，走真实 invoke_scoped 路径。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from langchain_core.language_models.chat_models import BaseChatModel
|
|
from langchain_core.messages import AIMessage
|
|
from langchain_core.outputs import ChatGeneration, ChatResult
|
|
|
|
from api import prescreen as prescreen_api
|
|
from api import project_assistant as api
|
|
from api import risk_agent_config as catalog
|
|
from contracts.prescreen import ErrorCode
|
|
from services import assistant_store as store
|
|
from services import knowledge_cards
|
|
from services.project_assistant import ProjectAssistantService
|
|
|
|
PROJECT = "DEMO-B"
|
|
PRESCREEN = prescreen_api.router.prefix
|
|
MANAGER = {"X-Xingbang-Demo-Role": "manager"}
|
|
REVIEWER = {"X-Xingbang-Demo-Role": "reviewer"}
|
|
|
|
CARD_ID = "skill.history.fixture"
|
|
|
|
|
|
class FakeModel(BaseChatModel):
|
|
"""假传输层；`calls` 记录调用次数，用于证明"确实发起了调用"。"""
|
|
|
|
payload: str = ""
|
|
calls: list = []
|
|
|
|
@property
|
|
def _llm_type(self) -> str:
|
|
return "fake"
|
|
|
|
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
|
|
FakeModel.calls.append(messages)
|
|
return ChatResult(generations=[ChatGeneration(message=AIMessage(content=self.payload))])
|
|
|
|
|
|
class GovernanceTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
FakeModel.calls = []
|
|
self.temp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
|
|
self.path = Path(self.temp.name)
|
|
self.service = ProjectAssistantService(self.path / "assistant.sqlite3")
|
|
self.patches = [
|
|
patch.object(api, "service", lambda: self.service),
|
|
patch.object(prescreen_api, "_service", lambda: self.service),
|
|
patch.object(catalog, "DB_PATH", self.path / "catalog.sqlite3"),
|
|
patch.dict(os.environ, {"XINGBANG_AUTH_USERS": "", "XINGBANG_REQUIRE_AUTH": "0",
|
|
"XINGBANG_LOCAL_MODEL_URL": "", "XINGBANG_DEMO_ROLE_LOGIN": "1"}),
|
|
]
|
|
for item in self.patches:
|
|
item.start()
|
|
app = FastAPI()
|
|
app.include_router(api.router)
|
|
app.include_router(prescreen_api.router)
|
|
self.client = TestClient(app)
|
|
with self.service.connect() as db:
|
|
store.put(db, "project", {
|
|
"id": PROJECT, "data_domain": "demo", "revision": 1,
|
|
"facts": {}, "values": {}, "sources": {}, "selected_sources": {},
|
|
"document_ids": [], "pending_candidates": [], "data_revision": 1,
|
|
})
|
|
|
|
def tearDown(self):
|
|
self.client.close()
|
|
for item in reversed(self.patches):
|
|
item.stop()
|
|
self.temp.cleanup()
|
|
|
|
# ------------------------------------------------------------ 夹具与工具
|
|
def propose_history_card(self, card_id: str = CARD_ID, actor: str = "demo-manager"):
|
|
with self.service.connect() as db:
|
|
return knowledge_cards.propose(db, {
|
|
"id": card_id, "kind": "skill", "name": "核查回款集中依赖",
|
|
"engine": "required", "inputs": ["financial.operating_cashflow"],
|
|
"code": "history.fixture", "title": "核查回款集中依赖",
|
|
"conditions": [{"code": "c1", "field": "collection.dependency",
|
|
"operator": "eq", "threshold": True}],
|
|
}, actor, PROJECT, origin="history_derived", source_event_ids=["event-1"])
|
|
|
|
def add_test_cases(self, card_id: str = CARD_ID, headers=MANAGER):
|
|
"""补独立测试样本——14 §4 的"专家修改 → 测试"步骤，也是发布门禁的前置条件。"""
|
|
return self.request("PATCH", f"/knowledge-candidates/{card_id}", {
|
|
"changes": {"test_cases": [
|
|
{"name": "缺证样本", "values": {"financial.operating_cashflow": None},
|
|
"expected_outputs": {}, "expected_checks": {"history.fixture": "unknown"}},
|
|
{"name": "已提供样本", "values": {"financial.operating_cashflow": "1200"},
|
|
"expected_outputs": {}, "expected_checks": {"history.fixture": "pass"}},
|
|
]},
|
|
"note": "补独立测试样本以满足评测门禁",
|
|
}, headers)
|
|
|
|
def card(self, card_id: str = CARD_ID) -> dict:
|
|
with self.service.connect() as db:
|
|
draft = knowledge_cards.current_draft(db, card_id)
|
|
return knowledge_cards.draft_card(db, draft["id"])
|
|
|
|
def request(self, method, path, body=None, headers=MANAGER):
|
|
return self.client.request(method, PRESCREEN + path, json=body, headers=headers)
|
|
|
|
def with_fake_model(self, payload):
|
|
return patch("services.experience_extraction.ChatOpenAI",
|
|
lambda **kwargs: FakeModel(payload=json.dumps(payload, ensure_ascii=False)))
|
|
|
|
|
|
class NaturalLanguageRevisionTests(GovernanceTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.propose_history_card()
|
|
|
|
def test_natural_language_revision_actually_calls_the_model(self):
|
|
answer = {"conditions": {"collection.dependency": False},
|
|
"exception_note": "仅适用于设备类租赁", "note": "限定适用范围"}
|
|
env = {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1",
|
|
"XINGBANG_LOCAL_MODEL_NAME": "fake-local"}
|
|
with patch.dict(os.environ, env), self.with_fake_model(answer):
|
|
response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}",
|
|
{"requirement": "只适用于设备类租赁，并加上例外"})
|
|
|
|
self.assertEqual(response.status_code, 200, response.text)
|
|
self.assertEqual(len(FakeModel.calls), 1, "自然语言修订必须真的发起一次模型调用")
|
|
card = self.card()
|
|
self.assertEqual(card["conditions"][0]["field"], "collection.dependency")
|
|
self.assertEqual(card["conditions"][0]["value"], False, "修订应真正落到卡片上")
|
|
|
|
def test_revision_keeps_history_origin_sources_and_draft_status(self):
|
|
with patch.dict(os.environ, {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}):
|
|
self.request("PATCH", f"/knowledge-candidates/{CARD_ID}",
|
|
{"changes": {"exception_note": "已有到账凭证时不重复核查"},
|
|
"note": "补充例外说明"})
|
|
card = self.card()
|
|
self.assertEqual(card["origin"], "history_derived", "人工修订不得洗掉历史来源")
|
|
self.assertEqual(card["source_event_ids"], ["event-1"])
|
|
self.assertEqual(card["status"], "draft", "修订只产草稿，不得因此生效")
|
|
self.assertGreater(card["revision"], 1, "修订必须产生新 revision")
|
|
|
|
def test_unavailable_model_blocks_natural_language_and_leaves_card_untouched(self):
|
|
before = self.card()
|
|
response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}",
|
|
{"requirement": "只适用于设备类租赁"})
|
|
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertIn("自然语言修改需要已配置模型", response.json()["error"]["message"])
|
|
after = self.card()
|
|
self.assertEqual(after["conditions"], before["conditions"], "拒绝时必须一字未改")
|
|
self.assertEqual(after["revision"], before["revision"])
|
|
|
|
def test_model_output_outside_standard_fields_is_rejected(self):
|
|
before = self.card()
|
|
with patch.dict(os.environ, {"XINGBANG_LOCAL_MODEL_URL": "http://127.0.0.1:11434/v1"}), \
|
|
self.with_fake_model({"conditions": {"自造字段": True}, "note": "乱改"}):
|
|
response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}",
|
|
{"requirement": "随便改"})
|
|
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertIn("必须来自标准字段", response.json()["error"]["message"])
|
|
self.assertEqual(self.card()["conditions"], before["conditions"])
|
|
|
|
def test_revision_requires_a_change_note(self):
|
|
response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}",
|
|
{"changes": {"exception_note": "x"}, "note": " "})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertIn("变更说明", response.json()["error"]["message"])
|
|
|
|
def test_revision_rejects_fields_outside_the_editable_set(self):
|
|
response = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}",
|
|
{"changes": {"scope": "production"}, "note": "改数据域"})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertIn("不支持的修订字段", response.json()["error"]["message"])
|
|
|
|
|
|
class PublicationTests(GovernanceTestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.propose_history_card()
|
|
|
|
def publish(self, headers=REVIEWER):
|
|
return self.request("POST", f"/knowledge-candidates/{CARD_ID}/transitions",
|
|
{"action": "publish"}, headers)
|
|
|
|
def test_publish_requires_an_independent_evaluation_first(self):
|
|
response = self.publish()
|
|
self.assertEqual(response.status_code, 409)
|
|
self.assertIn("评测", response.json()["error"]["message"])
|
|
|
|
def test_manager_cannot_evaluate_or_publish(self):
|
|
"""角色门禁与既有 /components 入口一致：客户经理不得自评自批。"""
|
|
evaluate = self.request("POST", f"/knowledge-candidates/{CARD_ID}/transitions",
|
|
{"action": "evaluate"})
|
|
self.assertEqual(evaluate.status_code, 403)
|
|
|
|
def test_self_approval_is_rejected(self):
|
|
"""AC-03/AC-43：提议人不能给自己的候选发布版本。"""
|
|
drafted = self.propose_history_card("skill.history.selfmade", actor="demo-reviewer")
|
|
evaluated = self.request("POST", f"/knowledge-candidates/{drafted['component_id']}/transitions",
|
|
{"action": "evaluate"}, REVIEWER)
|
|
self.assertEqual(evaluated.status_code, 200, evaluated.text)
|
|
published = self.request("POST", f"/knowledge-candidates/{drafted['component_id']}/transitions",
|
|
{"action": "publish"}, REVIEWER)
|
|
self.assertEqual(published.status_code, 403, "同一人不得既提议又发布")
|
|
|
|
def test_revision_after_publication_bumps_version_and_keeps_lineage(self):
|
|
self.assertEqual(self.add_test_cases().status_code, 200)
|
|
self.request("POST", f"/knowledge-candidates/{CARD_ID}/transitions", {"action": "evaluate"},
|
|
REVIEWER)
|
|
published = self.publish()
|
|
self.assertEqual(published.status_code, 200, published.text)
|
|
self.assertEqual(published.json()["version"], "1.0.0")
|
|
|
|
revised = self.request("PATCH", f"/knowledge-candidates/{CARD_ID}",
|
|
{"changes": {"exception_note": "补充例外"}, "note": "上会前修订"},
|
|
REVIEWER)
|
|
self.assertEqual(revised.status_code, 200, revised.text)
|
|
card = self.card()
|
|
self.assertEqual(card["version"], "1.0.1", "已发布版本不可覆盖，必须升版本")
|
|
self.assertEqual(card["parent_version"], {"id": CARD_ID, "version": "1.0.0"})
|
|
self.assertEqual(card["status"], "draft")
|
|
|
|
|
|
class UnknownObjectTests(GovernanceTestCase):
|
|
"""不存在的卡片必须是 404，不是 500。
|
|
|
|
这是回归守卫：归属查询若放在 run() 之外，KeyError 会绕过错误信封映射变成 500。
|
|
"""
|
|
|
|
def test_revising_an_unknown_candidate_is_404(self):
|
|
response = self.request("PATCH", "/knowledge-candidates/skill.not.exists",
|
|
{"changes": {"exception_note": "x"}, "note": "改"})
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertEqual(response.json()["error"]["code"], ErrorCode.NOT_FOUND.value)
|
|
|
|
def test_transitioning_an_unknown_candidate_is_404(self):
|
|
response = self.request("POST", "/knowledge-candidates/skill.not.exists/transitions",
|
|
{"action": "evaluate"}, REVIEWER)
|
|
self.assertEqual(response.status_code, 404)
|
|
|
|
|
|
class ManualCandidateTests(GovernanceTestCase):
|
|
def test_manual_candidate_is_created_as_draft(self):
|
|
response = self.request("POST", f"/projects/{PROJECT}/knowledge-candidates", {
|
|
"kind": "skill", "title": "人工新增：核对租赁物发票",
|
|
"content": {"engine": "required", "inputs": ["lease.total_value"],
|
|
"code": "manual.invoice"},
|
|
})
|
|
self.assertEqual(response.status_code, 201, response.text)
|
|
card = self.card(response.json()["component_id"])
|
|
self.assertEqual(card["origin"], "manual")
|
|
self.assertEqual(card["status"], "draft")
|
|
self.assertEqual(card["source_event_ids"], [])
|
|
|
|
def test_unknown_origin_is_rejected(self):
|
|
response = self.request("POST", f"/projects/{PROJECT}/knowledge-candidates", {
|
|
"kind": "skill", "title": "x", "origin": "模型自创",
|
|
"content": {"engine": "required", "inputs": ["lease.total_value"],
|
|
"code": "manual.x"},
|
|
})
|
|
self.assertEqual(response.status_code, 422)
|
|
self.assertIn("来源取值须为", response.json()["error"]["message"])
|
|
|
|
def test_candidate_list_exposes_origin_and_status(self):
|
|
self.propose_history_card()
|
|
listed = self.request("GET", f"/projects/{PROJECT}/knowledge-candidates").json()["items"]
|
|
self.assertEqual(len(listed), 1)
|
|
self.assertEqual(listed[0]["origin"], "history_derived")
|
|
self.assertEqual(listed[0]["status"], "draft")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|