"""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 registry_conditions(mapping: dict) -> list[dict]: """{标准字段: 触发值} → 注册表既有的等值条件形状。 注册表条件只能是 `{code, field, operator, threshold}` 且算子限于数值阈值， 因此这里固定用 `eq`——**不试图表达"字段存在/缺失"**，那种语义在注册表条件语言里 没有对应算子（`present` 被 test_knowledge_cards 的既有断言锁定拒绝）。 历史提炼与人工修订共用本函数，避免两处各写一份形状。 """ return [{"code": "history.condition." + str(index + 1), "field": str(field), "operator": "eq", "threshold": value} for index, (field, value) in enumerate(sorted((mapping or {}).items()))] def draft_card(db, draft_id: str) -> dict: return _project(store.get(db, "component_draft", draft_id)) # ---------------------------------------------------------------- 候选读取与修订（AC-51） def current_draft(db, card_ref: str) -> dict: """按草稿 id（`组件id@版本`）或组件 id 找当前候选草稿。 判据与方案草稿完全相同，因此直接复用注册表实现，不在这里再写一份排序规则。 """ return registry.current_draft(db, card_ref) def list_cards(db, project_id: str, domain: str | None = None) -> list[dict]: """项目下的知识卡片（草稿与已发布）。内置全局组件不是项目卡片，按 `_project` 的 既有判据排除，不在这里另写一套过滤。 """ cards = [] for kind in ("component_draft", "component_version"): for row in store.listing(db, kind, project_id): if row.get("project_id") != project_id: continue if domain and row.get("scope") != domain: continue try: cards.append(_project(row)) except ValueError: continue cards.sort(key=lambda card: (card["id"], card["status"] != "published", card["revision"])) return cards # 可改字段：适用范围、例外说明，以及**独立测试样本**。 # test_cases 在这里不是可选项——`component_evaluation` 对自定义 Skill 的发布门禁就是 # "必须有独立测试样本与期望输出"（14 §4 的"专家修改 → 测试"步骤）。不开放它， # 提炼出来的候选就永远卡在评测门外，闭环无法走通。 EDITABLE_CHANGES = ("conditions", "exception_note", "test_cases") def _bump(version: str) -> str: parts = str(version).split(".") if len(parts) == 3 and all(part.isdigit() for part in parts): return f"{parts[0]}.{parts[1]}.{int(parts[2]) + 1}" return f"{version}.1" def _redraft_body(row: dict) -> dict: """把已存在的草稿行还原成可再次提交的注册表体。 必须去掉 `registry.draft()` 自己维护或校验用的键：`revision`（否则会被当成 expected 与 `id` 拼接重复计版本）、`status`/`edited_by`/`approved_by` 等由服务端 决定的字段、以及 `content_hash`（会重算）。 """ body = {key: value for key, value in row.items() if key not in ("revision", "content_hash", "status", "edited_by", "approved_by", "published_at", "evaluation_id", "schema_version", "dependencies")} body["id"] = row.get("component_id") or row["id"] return body def revise(db, card_ref: str, changes: dict, actor: str, note: str, expected_revision=None): """按人工修改形成**草稿修订**（AC-51 / BR-42）。 三条不变量，全部由既有治理保证而不是本函数自己承诺： - **只产草稿**：走 `propose()` → `registry.draft()`，发布仍需独立审核与隔离评测； - **来源不可洗**：origin 与来源事件原样继承，历史卡片修订后仍是 `history_derived`； - **不自批**：本函数不发布；发布时 `registry.release()` 才检查 `edited_by == actor`。 已发布的当前版本会被**升版本**并记 `parent_version`；未发布时原地产生新 revision （与既有"草稿可反复修订"一致）。 """ if not str(note or "").strip(): raise ValueError("修订必须填写变更说明，否则无法追溯改了什么") unknown = sorted(set(changes or ()) - set(EDITABLE_CHANGES)) if unknown: raise ValueError(f"不支持的修订字段：{unknown}；本版只允许改适用范围、例外说明与独立测试样本") if not changes: raise ValueError("修订必须至少改动一个字段") draft = current_draft(db, card_ref) component_id = draft["component_id"] version_id = f"{component_id}@{draft['version']}" try: pointer = store.get(db, "release", component_id) except KeyError: pointer = None if pointer and pointer.get("version_id") == version_id: version = _bump(draft["version"]) parent = {"id": component_id, "version": draft["version"]} else: version = draft["version"] parent = draft.get("parent_version") body = _redraft_body(draft) body["version"] = version body["change_note"] = str(note) if "conditions" in changes: body["conditions"] = registry_conditions(changes["conditions"]) if "exception_note" in changes: body["exception_note"] = str(changes["exception_note"]) if "test_cases" in changes: cases = changes["test_cases"] if not isinstance(cases, list) or any( not isinstance(case, dict) or not str(case.get("name") or "").strip() for case in cases): raise ValueError("测试样本必须是带 name 的对象列表") body["test_cases"] = cases if expected_revision is not None: body["revision"] = expected_revision # registry.draft 用它做乐观并发校验 return propose(db, body, actor, draft["project_id"], origin=draft["origin"], source_event_ids=draft.get("source_event_ids"), source_evidence_ids=draft.get("source_evidence_ids"), parent_version=parent, card_scope=draft.get("card_scope", "project"), team_id=draft.get("team_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 # ---------------------------------------------------------------- AC-26 # 运行时的"本次预审补充说明"只影响**本次运行**（run_instruction 落在运行对象上， # 见 03 §9.3）。要把它变成持久知识，唯一途径是**显式提议为草稿**，且草稿走正常 # 审核流程；指令本身不得改变工具权限。 RUN_SCOPE_FIELDS = ("run_instruction", "tools", "allowed_tools", "permissions", "engine") def draft_from_run_instruction(db, project_id: str, run_id: str, body: dict, actor: str): """把一次运行的补充说明转成**待审草稿**（AC-26）。 三条约束： 1. 只产草稿，绝不因"运行中说过"而直接生效 2. 指令不得携带工具/权限/引擎字段——扩大工具授权必须走独立提议与审核 3. 保留来源运行，便于回溯"这条知识从哪次运行来" """ forbidden = sorted(field for field in RUN_SCOPE_FIELDS if field in body) if forbidden: raise ValueError( "本次预审补充说明不能持久化工具权限或执行引擎；" f"请移除这些字段并通过独立提议与审核：{forbidden}" ) if not body.get("name") or not body.get("kind"): raise ValueError("转为持久知识需要明确的名称与类型") origin = body.get("origin", "manual") if origin not in ORIGINS: raise ValueError(f"来源取值须为 {' 或 '.join(ORIGINS)}") # 运行指令天然没有历史来源，历史来源必须另行提供证据 if origin == "history_derived": raise ValueError("运行补充说明不能声明为历史提炼；历史提炼需提供来源事件或证据") enriched = dict(body, derived_from_run=run_id, proposed_via="run_instruction") return propose(db, enriched, actor, project_id, origin=origin, card_scope=body.get("card_scope", "project"), team_id=body.get("team_id"))