diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py index 2bca91d..0793323 100644 --- a/backend/contracts/prescreen.py +++ b/backend/contracts/prescreen.py @@ -415,3 +415,115 @@ class CaseTimeline(Contract): if self.status == "ready" and any(e.occurred_at is None for e in self.events): raise ValueError("存在缺发生时间的事件时状态不得为 ready") return self + + +# ---------------------------------------------------------------- 知识卡片(T05-b) +class Origin(StrEnum): + """知识创建来源,与 data_domain 正交(docs/v2/14-experience-loop.md §4)。""" + + history_derived = "history_derived" + manual = "manual" + policy_import = "policy_import" + + +class GovernanceStatus(StrEnum): + draft = "draft" + in_review = "in_review" + published = "published" + rejected = "rejected" + withdrawn = "withdrawn" + + +class RuleEffect(StrEnum): + advisory = "advisory" + hard = "hard" + + +class Condition(Contract): + """安全条件表达:只允许字段白名单 + 有限算子,不执行表达式(03 §9.1)。""" + + field: str = Field(min_length=1, max_length=ID_MAX) + operator: Literal["eq", "in", "present", "absent", "gt", "gte", "lt", "lte"] + value: str | bool | list[str] | None = None + + @model_validator(mode="after") + def _check_value_shape(self) -> Condition: + if self.operator == "in": + if not isinstance(self.value, list) or not self.value: + raise ValueError("in 算子需要非空列表") + elif self.operator in ("present", "absent"): + if self.value is not None: + raise ValueError("present/absent 不接受取值") + elif self.operator in ("gt", "gte", "lt", "lte"): + # 数值算子必须声明可解析的数值;比较口径由程序校验,不做表达式求值 + try: + float(str(self.value)) + except (TypeError, ValueError): + raise ValueError("数值算子需要可解析的数值") from None + return self + + +class RuleContent(Contract): + """规则内容。hard 必须具备正式依据与独立审核(BR-26 / E-18)。""" + + effect: RuleEffect + predicate: list[Condition] = Field(min_length=1, max_length=32) + on_match: Literal["flag", "block"] + basis_evidence_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX) + policy_ref: Ref | None = None + authorization_ref: Ref | None = None + + @model_validator(mode="after") + def _check_hard_basis(self) -> RuleContent: + if self.effect == RuleEffect.hard: + if self.policy_ref is None and self.authorization_ref is None: + raise ValueError("硬性规则需要正式制度或授权依据,仅凭历史频次不足") + if not self.basis_evidence_ids: + raise ValueError("硬性规则必须附依据证据") + return self + + +class KnowledgeCard(Contract): + """Skill/Rule 卡片的目标投影。 + + **命名冲突提示**:本类的 `card_scope` 是 project/team(03 §9.2),而既有 + `component_registry` 里的 `scope` 表示数据域 demo/production。二者语义不同, + 投影层必须显式映射,不得互相顶替。 + """ + + id: Identifier + project_id: Identifier + data_domain: DataDomain + revision: int = Field(ge=1) + version: str = Field(min_length=1, max_length=70) + kind: str = Field(min_length=1, max_length=40) + title: str = Field(min_length=1, max_length=200) + origin: Origin + status: GovernanceStatus + source_event_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX) + source_evidence_ids: list[Identifier] = Field(default_factory=list, max_length=EVIDENCE_MAX) + parent_version: Ref | None = None + created_by: Identifier + edited_by: Identifier + card_scope: Literal["project", "team"] + team_id: str | None = Field(default=None, max_length=ID_MAX) + conditions: list[Condition] = Field(default_factory=list, max_length=32) + exceptions: list[Condition] = Field(default_factory=list, max_length=32) + effective_at: AwareDatetime + expires_at: AwareDatetime | None = None + reviewer_id: Identifier | None = None + evaluation_ref: Ref | None = None + rule: RuleContent | None = None + + @model_validator(mode="after") + def _check_card(self) -> KnowledgeCard: + if self.card_scope == "team" and not self.team_id: + raise ValueError("team 范围的卡片必须有 team_id") + if self.origin == Origin.history_derived and not ( + self.source_event_ids or self.source_evidence_ids): + raise ValueError("历史提炼卡片必须保留来源(事件或证据)") + if self.status == GovernanceStatus.published and self.reviewer_id is None: + raise ValueError("已发布卡片必须有审核人") + if self.effective_at and self.expires_at and self.expires_at <= self.effective_at: + raise ValueError("有效期必须晚于生效时间") + return self diff --git a/backend/services/knowledge_cards.py b/backend/services/knowledge_cards.py new file mode 100644 index 0000000..dc48c54 --- /dev/null +++ b/backend/services/knowledge_cards.py @@ -0,0 +1,198 @@ +"""T05-b 知识卡片:既有 Skill/Rule 治理的目标投影。 + +**不新建第二套发布库**(docs/v2/03-detailed-design.md §9.2 明文)。本模块只负责三件事: + +1. **投影字段**:origin、谱系(parent_version)、数据域、卡片范围(project/team)、条件与例外 +2. **硬规则门禁**:hard 必须有正式制度/授权依据与依据证据(BR-26 / E-18) +3. **有效性过滤**:按数据域 / 状态 / 有效期 / 范围筛选可应用的已发布版本(BR-04) + +状态机与发布**完全委托** `component_registry.draft()` / `release()`,因此既有保证 +原样生效、不重复实现: +- 自批拒绝:`release()` 里 `edited_by == actor` → PermissionError(AC-03 / AC-43) +- 评测门禁:候选哈希必须匹配通过的隔离评测(AC-34) +- 已发布不可覆盖:`component_version` 以 immutable 写入;改已发布需新版本号 +- 工具白名单、方案预算等既有校验 + +**命名映射(不得互相顶替)**: +- 本设计的 `card_scope`(project/team)↔ 注册表无此概念,作为附加键保存 +- 本设计的 `data_domain` ↔ 注册表的 `scope`(demo/production)——**同一个键名、不同语义**, + 投影时显式互换,绝不混用 +""" + +from __future__ import annotations + +from contracts.prescreen import Condition, KnowledgeCard, Origin, RuleEffect +from services import assistant_store as store +from services import component_registry as registry + +ORIGINS = tuple(item.value for item in Origin) +EFFECTS = tuple(item.value for item in RuleEffect) + +HARD_BASIS_REQUIRED = "硬性规则需要正式制度或授权依据,仅凭历史频次不足;当前仅可保存为建议草稿" + + +def _check_hard_basis(body: dict) -> None: + """hard 规则的资格校验(BR-26 / E-18)。advisory 不设门槛。""" + if body.get("kind") != "rule_set": + return + effect = body.get("effect", "advisory") + if effect not in EFFECTS: + raise ValueError(f"规则效果取值须为 {' 或 '.join(EFFECTS)}") + if effect != "hard": + return + if not (body.get("policy_ref") or body.get("authorization_ref")): + raise ValueError(HARD_BASIS_REQUIRED) + if not body.get("basis_evidence_ids"): + raise ValueError("硬性规则必须附依据证据,不能只有结论") + + +# 注册表既有的条件形状与本设计的 Condition **不是同一套**: +# 注册表:{code, field, operator, threshold, title, suggestion} +# 本设计:{field, operator, value} +# 因此投影层做**翻译**,而不是把一方强加给另一方——注册表体按它自己的形状保存, +# 卡片侧只暴露可安全求值的字段,展示性字段(title/suggestion)不进契约。 +# 只映射注册表**实际可能产出**的算子:component_registry.validate() 要求 +# operator ∈ (gt, lt, eq, gte, lte) 且 threshold 可解析为数值。 +# 本设计的 Condition 还允许 in/present/absent,但它们不可能来自注册表条件, +# 故不在此表中——出现即报错,避免把不可能的取值当成已支持。 +_OPERATORS = {"eq": "eq", "lt": "lt", "lte": "lte", "gt": "gt", "gte": "gte"} + + +def _translate_conditions(conditions, label: str) -> list[dict]: + """把注册表条件翻译为设计的安全条件;无法表达即报错,不静默丢弃。""" + parsed = [] + for index, item in enumerate(conditions or ()): + if not isinstance(item, dict): + raise ValueError(f"{label}第 {index + 1} 条不是对象") + operator = _OPERATORS.get(str(item.get("operator", "")).strip().lower()) + if operator is None: + raise ValueError(f"{label}第 {index + 1} 条算子不受支持:{item.get('operator')!r}") + raw = item.get("threshold", item.get("value")) + value = raw if isinstance(raw, (list, bool)) else (None if raw is None else str(raw)) + try: + parsed.append(Condition(field=str(item.get("field", "")), operator=operator, + value=value).model_dump(mode="json")) + except Exception as exc: + raise ValueError(f"{label}第 {index + 1} 条不合法:{exc}") from exc + if not parsed and conditions: + raise ValueError(f"{label}无法表达为安全条件") + return parsed + + +def propose(db, body: dict, actor: str, project_id: str, origin: str = "manual", + source_event_ids=(), source_evidence_ids=(), parent_version: dict | None = None, + card_scope: str = "project", team_id: str | None = None): + """提议一张知识卡片草稿。**只生成草稿,绝不直接生效**(BR-03 / AC-26)。""" + if origin not in ORIGINS: + raise ValueError(f"来源取值须为 {' 或 '.join(ORIGINS)}") + if card_scope not in ("project", "team"): + raise ValueError("卡片范围须为 project 或 team") + if card_scope == "team" and not team_id: + raise ValueError("team 范围的卡片必须有 team_id") + if origin == "history_derived" and not (source_event_ids or source_evidence_ids): + raise ValueError("历史提炼卡片必须保留来源(事件或证据),不得成为无源知识") + if "data_domain" in body: + raise ValueError("请用 scope 声明数据域(注册表既有约定),不要另传 data_domain") + + _check_hard_basis(body) + enriched = dict( + body, + origin=origin, + project_id=project_id, + card_scope=card_scope, + team_id=team_id, + parent_version=parent_version, + source_event_ids=list(source_event_ids), + source_evidence_ids=list(source_evidence_ids), + ) + # 只校验"能否表达为安全条件",不改写注册表既有的条件形状 + _translate_conditions(body.get("conditions"), "适用条件") + _translate_conditions(body.get("exceptions"), "例外") + # 委托既有治理:normalize + validate + 写入 component_draft + return registry.draft(db, enriched, actor) + + +def publish(db, draft_id: str, actor: str, preview: bool = False): + """发布草稿。自批与评测门禁由 `component_registry.release()` 保证。""" + body = store.get(db, "component_draft", draft_id) + # 门禁在发布时**再查一次**:草稿可能在提议之后被改动,不能只信提议时的校验 + _check_hard_basis(body) + return registry.release(db, draft_id, actor, preview) + + +def _project(body: dict) -> dict: + """把注册表对象投影为 KnowledgeCard。scope↔data_domain 在此完成互换。""" + domain = body.get("scope") + if domain not in ("demo", "production"): + raise ValueError("组件的数据域不合法,无法投影为卡片") + # 内置种子组件是**全局**的(无 project_id / edited_by),不是项目知识卡片。 + # 明确报错而不是让 pydantic 抛出难懂的信息——否则容易把内置 Skill 误当项目卡片。 + if not body.get("project_id") or not body.get("edited_by"): + raise ValueError( + "该组件不是项目知识卡片(缺少 project_id 或 edited_by);" + "内置 Skill/规则集属全局配置,请勿投影为卡片" + ) + card = { + "id": body.get("component_id") or body["id"], + "project_id": body.get("project_id") or "", + "data_domain": domain, # ← 注册表 scope + "revision": body.get("revision", 1), + "version": body.get("version", "1.0.0"), + "kind": body.get("kind", ""), + "title": body.get("name") or body.get("title") or "", + "origin": body.get("origin", "manual"), + "status": body.get("status", "draft"), + "source_event_ids": body.get("source_event_ids", []), + "source_evidence_ids": body.get("source_evidence_ids", []), + "parent_version": body.get("parent_version"), + "created_by": body.get("owner") or body.get("edited_by", ""), + "edited_by": body.get("edited_by", ""), + "card_scope": body.get("card_scope", "project"), + "team_id": body.get("team_id"), + "conditions": _translate_conditions(body.get("conditions"), "适用条件"), + "exceptions": _translate_conditions(body.get("exceptions"), "例外"), + "effective_at": body.get("effective_from") or store.now(), + "expires_at": body.get("expires_at"), + "reviewer_id": body.get("approved_by"), + "rule": None, + } + if body.get("kind") == "rule_set": + card["rule"] = { + "effect": body.get("effect", "advisory"), + "predicate": _translate_conditions( + body.get("predicate") or body.get("conditions"), "规则条件") or [ + {"field": "unknown", "operator": "present"}], + "on_match": body.get("on_match", "flag"), + "basis_evidence_ids": body.get("basis_evidence_ids", []), + "policy_ref": body.get("policy_ref"), + "authorization_ref": body.get("authorization_ref"), + } + return KnowledgeCard(**card).model_dump(mode="json") + + +def draft_card(db, draft_id: str) -> dict: + return _project(store.get(db, "component_draft", draft_id)) + + +def published_cards(db, domain: str) -> list[dict]: + return [_project(row) for row in store.listing(db, "component_version") + if row.get("status") == "published" and row.get("scope") == domain] + + +def effective(db, domain: str, project_id: str | None = None, at: str | None = None) -> list[dict]: + """可应用的已发布卡片(BR-04:先按数据域/状态/有效期/范围过滤)。 + + 用户级权限(OQ-02)尚未建设,故此处只做域、有效期与范围过滤; + **不得**据此宣称已完成授权过滤。 + """ + moment = at or store.now() + result = [] + for card in published_cards(db, domain): + if card["expires_at"] and card["expires_at"] <= moment: + continue + if card["card_scope"] == "project" and project_id and card["project_id"] != project_id: + continue + if card["card_scope"] == "team" and not card["team_id"]: + continue + result.append(card) + return result diff --git a/backend/test_knowledge_cards.py b/backend/test_knowledge_cards.py new file mode 100644 index 0000000..deb001f --- /dev/null +++ b/backend/test_knowledge_cards.py @@ -0,0 +1,224 @@ +"""T05-b 知识卡片治理测试:AC-03 / AC-25 / AC-27 / AC-43。 + +重点证明两件事: +1. **没有第二套发布库**:发布路径完全走既有 component_registry,因此自批拒绝、 + 评测门禁、已发布不可覆盖这些既有保证对新卡片同样生效。 +2. **硬规则有正式依据门槛**:只有历史频次的候选不得发布为 hard(E-18)。 +""" + +import tempfile +import unittest +from pathlib import Path + +from services import assistant_store as store +from services import component_registry as registry +from services import knowledge_cards as cards + + +def rule_body(**over): + body = { + "id": "corporate.test_rule", "kind": "rule_set", "name": "测试规则", + "scope": "demo", "version": "1.0.0", + # 注册表规则条件的形状:operator ∈ (gt,lt,eq,gte,lte) 且 threshold 可解析为数值, + # title/suggestion 是展示字段(不进设计契约) + "conditions": [{"field": "metrics.coverage", "operator": "lt", "threshold": "1.2", + "title": "覆盖不足", "suggestion": "请补充担保措施"}], + "effect": "advisory", "on_match": "flag", + } + body.update(over) + return body + + +class KnowledgeCardTestCase(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) + self.db_path = Path(self._tmp.name) / "cards.sqlite3" + store.initialize(self.db_path) + + def tearDown(self): + self._tmp.cleanup() + + def propose(self, body=None, origin="manual", **kwargs): + with store.transaction(self.db_path) as db: + return cards.propose(db, body or rule_body(), "demo-manager", "DEMO-A", + origin=origin, **kwargs) + + def publish(self, draft_id, actor, preview=False): + with store.transaction(self.db_path) as db: + return cards.publish(db, draft_id, actor, preview) + + +class HardRuleGateTests(KnowledgeCardTestCase): + def test_frequency_only_candidate_cannot_be_hard(self): + # AC-27:只有历史频次,没有正式依据 → 拒绝 hard + with self.assertRaises(ValueError) as ctx: + self.propose(rule_body(effect="hard")) + self.assertIn("正式制度或授权依据", str(ctx.exception)) + self.assertIn("仅可保存为建议草稿", str(ctx.exception)) + + def test_frequency_only_candidate_can_stay_advisory(self): + draft = self.propose(rule_body(effect="advisory")) + self.assertEqual(draft["status"], "draft") + self.assertEqual(draft["effect"], "advisory") + + def test_hard_with_policy_reference_is_accepted(self): + draft = self.propose(rule_body( + effect="hard", on_match="block", + policy_ref={"id": "policy.credit-2025", "version": "v3"}, + basis_evidence_ids=["ev-1"])) + self.assertEqual(draft["effect"], "hard") + + def test_hard_requires_basis_evidence_too(self): + with self.assertRaises(ValueError) as ctx: + self.propose(rule_body(effect="hard", + policy_ref={"id": "p", "version": "v1"})) + self.assertIn("依据证据", str(ctx.exception)) + + def test_gate_is_rechecked_at_publish_time(self): + # 草稿在提议后被改成 hard(模拟旁路),发布时必须再查一次 + draft = self.propose(rule_body(effect="advisory")) + with store.transaction(self.db_path) as db: + row = store.get(db, "component_draft", draft["id"]) + row["effect"] = "hard" + store.put(db, "component_draft", row) + with self.assertRaises(ValueError) as ctx: + self.publish(draft["id"], "demo-reviewer") + self.assertIn("正式制度或授权依据", str(ctx.exception)) + + def test_unknown_effect_is_rejected(self): + with self.assertRaises(ValueError): + self.propose(rule_body(effect="mandatory")) + + def test_conditions_are_validated(self): + with self.assertRaises(ValueError): + self.propose(rule_body(conditions=[{"field": "f", "operator": "present"}])) + + +class OriginAndLineageTests(KnowledgeCardTestCase): + def test_history_derived_requires_sources(self): + with self.assertRaises(ValueError) as ctx: + self.propose(rule_body(), origin="history_derived") + self.assertIn("必须保留来源", str(ctx.exception)) + + def test_history_derived_keeps_lineage_after_manual_edit(self): + # AC-25:历史提炼卡被人工编辑后仍是历史来源,并保留父版本 + draft = self.propose(rule_body(id="corporate.from_history"), origin="history_derived", + source_event_ids=["event-1"]) + edited = self.propose(rule_body(id="corporate.from_history", version="1.0.1"), + origin="history_derived", source_event_ids=["event-1"], + parent_version={"id": "corporate.from_history", "version": "1.0.0"}) + with store.transaction(self.db_path) as db: + card = cards.draft_card(db, edited["id"]) + self.assertEqual(card["origin"], "history_derived") + self.assertEqual(card["parent_version"], {"id": "corporate.from_history", "version": "1.0.0"}) + self.assertEqual(card["source_event_ids"], ["event-1"]) + self.assertNotEqual(draft["id"], edited["id"]) + + def test_manual_card_has_manual_origin_and_no_history_sources(self): + with store.transaction(self.db_path) as db: + card = cards.draft_card(db, self.propose(rule_body(id="corporate.by_hand"))["id"]) + self.assertEqual(card["origin"], "manual") + self.assertEqual(card["source_event_ids"], []) + + def test_data_domain_maps_to_registry_scope_not_card_scope(self): + # 两个 scope 语义不同:data_domain 落注册表 scope,卡片范围落 card_scope + with store.transaction(self.db_path) as db: + card = cards.draft_card(db, self.propose( + rule_body(id="corporate.scoped", scope="production"), + card_scope="team", team_id="team-1")["id"]) + self.assertEqual(card["data_domain"], "production") + self.assertEqual(card["card_scope"], "team") + self.assertEqual(card["team_id"], "team-1") + + def test_team_scope_requires_team_id(self): + with self.assertRaises(ValueError): + self.propose(rule_body(), card_scope="team") + + def test_callers_cannot_smuggle_data_domain_key(self): + with self.assertRaises(ValueError) as ctx: + self.propose(rule_body(data_domain="production")) + self.assertIn("不要另传 data_domain", str(ctx.exception)) + + +class GovernanceReuseTests(KnowledgeCardTestCase): + def test_self_approval_is_rejected_by_existing_governance(self): + # AC-03 / AC-43:提议人不得自批——沿用既有 release() 的守卫 + draft = self.propose() + with self.assertRaises(PermissionError) as ctx: + self.publish(draft["id"], "demo-manager") + self.assertIn("独立审核人", str(ctx.exception)) + + def test_publish_still_requires_a_passing_evaluation(self): + # 既有评测门禁对新卡片同样生效:没有评测就不能发布 + draft = self.propose() + with store.transaction(self.db_path) as db: + with self.assertRaises(store.Conflict) as ctx: + cards.publish(db, draft["id"], "demo-reviewer") + self.assertIn("评测", str(ctx.exception)) + + def test_reproposing_the_same_version_updates_the_same_draft(self): + # 草稿可反复修订:同 id@version 仍是同一草稿,不会产生第二份 + first = self.propose(rule_body(id="corporate.versioned")) + second = self.propose(rule_body(id="corporate.versioned", name="改过的名字")) + self.assertEqual(first["id"], second["id"]) + self.assertEqual(second["name"], "改过的名字") + with store.transaction(self.db_path) as db: + self.assertEqual(len([r for r in store.listing(db, "component_draft") + if r.get("component_id") == "corporate.versioned"]), 1) + + def test_published_version_cannot_be_redrafted_at_the_same_version(self): + # AC-43:已发布编辑必须用新版本号,否则被既有治理拒绝 + self.propose(rule_body(id="corporate.versioned")) + with store.transaction(self.db_path) as db: + draft = store.get(db, "component_draft", "corporate.versioned@1.0.0") + row = dict(draft, status="published", approved_by="demo-reviewer", + component_id="corporate.versioned") + row.pop("revision", None) + store.put(db, "component_version", row, immutable=True) + with self.assertRaises(store.Conflict) as ctx: + self.propose(rule_body(id="corporate.versioned")) + self.assertIn("请使用新版本号", str(ctx.exception)) + # 换版本号即可继续演进 + bumped = self.propose(rule_body(id="corporate.versioned", version="1.0.1")) + self.assertTrue(bumped["id"].endswith("@1.0.1")) + + def test_propose_never_publishes_directly(self): + draft = self.propose() + self.assertEqual(draft["status"], "draft") + with store.transaction(self.db_path) as db: + published = [row for row in store.listing(db, "component_version") + if row.get("component_id") == "corporate.test_rule"] + self.assertEqual(published, [], "提议不得直接产生已发布版本") + + +class EffectiveFilterTests(KnowledgeCardTestCase): + def test_only_published_and_matching_domain_are_returned(self): + self.propose(rule_body(id="corporate.demo_only", scope="demo")) + self.propose(rule_body(id="corporate.prod_only", scope="production")) + with store.transaction(self.db_path) as db: + # 未发布 → 不返回 + self.assertEqual(cards.effective(db, "demo"), []) + # 手工把草稿标记为已发布以验证域过滤(发布路径本身由上面几个类覆盖) + for draft in store.listing(db, "component_draft"): + row = dict(draft, status="published", approved_by="demo-reviewer", + component_id=draft.get("component_id")) + row.pop("revision", None) + store.put(db, "component_version", row, immutable=True) + demo = cards.effective(db, "demo") + production = cards.effective(db, "production") + self.assertTrue(demo) + self.assertTrue(all(card["data_domain"] == "demo" for card in demo)) + self.assertTrue(all(card["data_domain"] == "production" for card in production)) + + def test_global_seed_components_are_not_mistakable_for_project_cards(self): + """内置种子组件是全局配置,不是项目知识卡片——必须明确报错而不是含糊通过。""" + for item in registry.seed_components(): + body = registry.normalize(dict(item, status="published"), "system-demo") + with self.subTest(component=item["id"]): + with self.assertRaises(ValueError) as ctx: + cards._project(body) + self.assertIn("不是项目知识卡片", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main()