# ============================================================ # 约束配置中心（moduleId: domain-constraints, 可重生 ✅） # SC-04：ConstraintDef 注册表 + ConstraintProfile；硬约束拦发布、软约束报风险 # ============================================================ from __future__ import annotations from copy import deepcopy from typing import Any World = dict[str, Any] # 冲突类型 → 约束 ID（发布门禁与启停映射） CONFLICT_TO_CONSTRAINT: dict[str, str] = { "NO_LINE": "C5_capability", "NO_WORKSTATION": "C5_capability", "NO_CAPABILITY": "C5_capability", "NO_ROUTING": "C5_capability", "NO_MOLD": "C12_tooling", "NO_TEAM": "C12_team", "MOLD_LIFE": "C12_tooling", "MATERIAL_SHORTAGE": "C6_material_kit", "EQUIPMENT": "C4_maintenance", "CAPACITY": "C7_capacity", "DELAY": "C8_due_date", "WINDOW_DEFERRED": "C11_freeze", "WINDOW_TRUNCATED": "C11_freeze", "SOP_CHANGEOVER_CURFEW": "C13_sop", } # 默认约束目录（plan §9.2a 子集；configurable=False 的项不可在配置中心关闭） DEFAULT_CONSTRAINT_DEFS: list[dict[str, Any]] = [ { "id": "C1_precedence", "code": "C1", "name": "工艺先后序", "group": "process", "kind": "hard", "enabled": True, "configurable": False, "description": "前序工序完成后才能开下一工序；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c1-precedence", }, { "id": "C2_no_overlap", "code": "C2", "name": "工位/设备独占", "group": "capacity", "kind": "hard", "enabled": True, "configurable": False, "description": "同一工位同一时刻不可重叠；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c2-no-overlap", }, { "id": "C3_calendar", "code": "C3", "name": "班次日历", "group": "calendar", "kind": "hard", "enabled": True, "configurable": False, "description": "工单只能落在可用班次内；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c3-calendar", }, { "id": "C4_maintenance", "code": "C4", "name": "设备维保窗口", "group": "equipment", "kind": "hard", "enabled": True, "configurable": True, "description": "维保时段不可排产；硬约束，违反报 EQUIPMENT 并阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c4-maintenance", }, { "id": "C5_capability", "code": "C5", "name": "机器资格/选线", "group": "equipment", "kind": "hard", "enabled": True, "configurable": False, "description": "工序只能落在有资格的产线/工位/能力池；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c5-capability", }, { "id": "C6_material_kit", "code": "C6", "name": "物料齐套", "group": "material", "kind": "soft", "enabled": True, "configurable": True, "kindConfigurable": True, "description": "缺料：硬=阻断发布；软=带风险排入；关闭=不检查（硬/软可配）", "weight": 0.8, "goldenCase": "golden-c6-material-kit", "params": {"strategy": "risk"}, }, { "id": "C7_capacity", "code": "C7", "name": "产线日产能上限", "group": "capacity", "kind": "hard", "enabled": True, "configurable": True, "description": "单日占用超出可用分钟则报 CAPACITY；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c7-capacity", }, { "id": "C8_due_date", "code": "C8", "name": "订单交期", "group": "order", "kind": "soft", "enabled": True, "configurable": True, "description": "延期进冲突与 KPI，默认不拦发布", "weight": 0.6, "goldenCase": "golden-c8-due-date", }, { "id": "C9_customer_priority", "code": "C9", "name": "客户优先级/VIP", "group": "order", "kind": "soft", "enabled": True, "configurable": True, "description": "客户等级与订单优先级的排序/延期权重来源（level_weight）", "weight": 0.8, "goldenCase": "golden-c9-customer-priority", }, { "id": "C10_changeover", "code": "C10", "name": "顺序相关换型", "group": "process", "kind": "soft", "enabled": True, "configurable": True, "description": "相邻工单产品族切换时叠加换型矩阵分钟（MD-06）", "weight": 0.7, "goldenCase": "golden-c10-changeover", }, { "id": "C11_freeze", "code": "C11", "name": "滚动/冻结窗口", "group": "order", "kind": "hard", "enabled": True, "configurable": True, "description": "窗外工单延期/截断（柔性滚动窗）；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c11-freeze", }, { "id": "C12_team", "code": "C12", "name": "班组人力并发", "group": "personnel", "kind": "hard", "enabled": True, "configurable": True, "description": "柔性轨班组同时占用上限（SC-11）；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c12-team", }, { "id": "C12_tooling", "code": "C12", "name": "工装/模具", "group": "tooling", "kind": "hard", "enabled": True, "configurable": True, "description": "模具适配与寿命锁定（SC-10）；硬约束，违反阻止发布，不可关闭", "weight": 1.0, "goldenCase": "golden-c12-tooling", }, { "id": "C13_sop", "code": "C13", "name": "SOP 规则包", "group": "process", "kind": "soft", "enabled": False, "configurable": True, "kindConfigurable": True, "description": "行业 SOP 编译生效标记（IND-02）；含换型宵禁等派生规则（硬/软可配）", "weight": 0.5, "goldenCase": "golden-c13-sop", "params": {}, }, ] GROUP_LABELS = { "process": "工艺", "capacity": "产能", "calendar": "日历", "equipment": "设备", "material": "物料", "order": "订单", "personnel": "人员", "tooling": "工装", } # plan.md §9.2a.1 的 13 类规范化类名（矩阵 113 盘点口径；C12 聚合班组/工装两个子资源） CLASS_NAMES: dict[str, str] = { "C1": "工艺先后序", "C2": "工位/设备独占", "C3": "班次日历", "C4": "设备维保窗口", "C5": "机器资格/选线", "C6": "物料齐套", "C7": "产线日产能上限", "C8": "订单交期", "C9": "客户优先级/VIP", "C10": "顺序相关换型", "C11": "滚动/冻结窗口", "C12": "班组/工装等累积资源", "C13": "企业SOP规则", } def _index_defaults() -> dict[str, dict[str, Any]]: return {d["id"]: deepcopy(d) for d in DEFAULT_CONSTRAINT_DEFS} def get_constraint_profile(world: World) -> dict[str, Any]: """合并默认目录与世界覆盖后的约束配置剖面。""" defaults = _index_defaults() stored = (world.get("constraintProfile") or {}).get("constraints") or {} items: list[dict[str, Any]] = [] for cid, base in defaults.items(): overlay = stored.get(cid) or {} item = deepcopy(base) if item.get("configurable"): if "enabled" in overlay: item["enabled"] = bool(overlay["enabled"]) if overlay.get("kind") in ("hard", "soft"): item["kind"] = overlay["kind"] if "weight" in overlay: try: item["weight"] = float(overlay["weight"]) except (TypeError, ValueError): pass if isinstance(overlay.get("params"), dict): item["params"] = {**(item.get("params") or {}), **overlay["params"]} items.append(item) return { "profileId": (world.get("constraintProfile") or {}).get("profileId") or "default", "name": (world.get("constraintProfile") or {}).get("name") or "默认约束剖面", "constraints": items, "groups": [ {"id": gid, "label": label, "items": [c["id"] for c in items if c["group"] == gid]} for gid, label in GROUP_LABELS.items() if any(c["group"] == gid for c in items) ], } def get_constraint(world: World, constraint_id: str) -> dict[str, Any] | None: for c in get_constraint_profile(world)["constraints"]: if c["id"] == constraint_id: return c return None def is_enabled(world: World, constraint_id: str) -> bool: c = get_constraint(world, constraint_id) return bool(c and c.get("enabled", True)) def constraint_kind(world: World, constraint_id: str) -> str: c = get_constraint(world, constraint_id) return str((c or {}).get("kind") or "hard") def engine_constraint_flags(world: World) -> dict[str, bool]: """映射到 EngineParams.constraints 布尔开关（兼容旧字段）。""" return { "materialKit": is_enabled(world, "C6_material_kit"), "equipment": is_enabled(world, "C4_maintenance"), "personnel": is_enabled(world, "C12_team"), "changeover": is_enabled(world, "C10_changeover"), "capacity": is_enabled(world, "C7_capacity"), "dueDate": is_enabled(world, "C8_due_date"), "tooling": is_enabled(world, "C12_tooling"), "freeze": is_enabled(world, "C11_freeze"), } def material_shortage_severity(world: World) -> str | None: """None=不检查；否则返回冲突 severity。""" if not is_enabled(world, "C6_material_kit"): return None return "MAJOR" if constraint_kind(world, "C6_material_kit") == "hard" else "MINOR" def profile_snapshot(world: World) -> dict[str, Any]: """写入排产版本的约束快照（只影响审计/解释，不回写配置）。""" p = get_constraint_profile(world) return { "profileId": p["profileId"], "constraints": { c["id"]: {"enabled": c["enabled"], "kind": c["kind"], "weight": c.get("weight", 1.0)} for c in p["constraints"] }, } def hard_blocking_conflicts(world: World, version_id: int | None = None, track: str = "fixed") -> list[dict[str, Any]]: """未解决且映射到「启用的硬约束」的冲突（发布门禁用）。""" key = "flexConflicts" if track == "flex" else "conflicts" rows = [] for c in world.get(key) or []: if c.get("isResolved") or c.get("status") == "RESOLVED": continue if version_id is not None and c.get("versionId") != version_id: continue cid = CONFLICT_TO_CONSTRAINT.get(str(c.get("conflictType") or "")) if not cid: continue if not is_enabled(world, cid): continue if constraint_kind(world, cid) != "hard": continue rows.append({**c, "constraintId": cid}) return rows class ConstraintProfileDenied(ValueError): """约束剖面保存被门禁拒绝（关闭硬约束/降级/软权重归零等）。 message 为中文说明；constraint_id/requested 供审计事件引用。 继承 ValueError：现有调用方（workflow/app 的 except ValueError 路径）无需改动。 """ def __init__(self, message: str, *, constraint_id: str | None = None, requested: dict[str, Any] | None = None) -> None: super().__init__(message) self.constraint_id = constraint_id self.requested = requested def constraint_catalog(world: World | None = None) -> dict[str, Any]: """只读 13 类约束目录（矩阵 113：类别齐全 + 每类硬/软 + 可关闭性 + 黄金用例标识）。 按 plan.md §9.2a.1 的 C1..C13 分类 code 聚合（C12 聚合班组/工装两个子资源）； 软约束可关闭（enabled=False），硬约束不可关闭（关闭/降级/软权重归零会被门禁拒绝）。 """ profile = get_constraint_profile(world if world is not None else {}) by_code: dict[str, dict[str, Any]] = {} for c in profile["constraints"]: code = c["code"] cls = by_code.setdefault(code, { "code": code, "name": CLASS_NAMES.get(code, c["name"]), "group": c.get("group"), "kind": "hard", "closable": True, "configurable": bool(c.get("configurable")), "kindConfigurable": bool(c.get("kindConfigurable")), "enabled": True, "goldenCase": c.get("goldenCase") or f"golden-{c['id']}", "items": [], }) cls["items"].append(c["id"]) if c["kind"] != "hard": cls["kind"] = "soft" else: cls["closable"] = False if not c.get("enabled", True): cls["enabled"] = False classes = [by_code[k] for k in sorted(by_code)] return { "catalogVersion": "9.2a.1", "total": len(classes), "classes": classes, "summary": { "hardCount": sum(1 for x in classes if x["kind"] == "hard"), "softCount": sum(1 for x in classes if x["kind"] == "soft"), "closableCount": sum(1 for x in classes if x["closable"]), "enabledCount": sum(1 for x in classes if x["enabled"]), }, } def _audit_next_id_for(world: World): """world 内 append-only 审计 ID 分配器（与 WorldStore.next_id('audit') 一致：max+1）。""" def _next_id(kind: str) -> int: ids = [e.get("id") for e in world.get("auditEvents", []) if isinstance(e.get("id"), int)] return (max(ids, default=0)) + 1 return _next_id def _write_profile_denied_audit(world: World, payload: dict[str, Any], exc: ConstraintProfileDenied) -> dict[str, Any]: """拒绝「关闭硬约束」类请求的审计事件（action=constraint.profile.save, result=DENIED）。""" from server.agent_core.audit import write_audit profile = world.get("constraintProfile") or {} return write_audit( world, _audit_next_id_for(world), actor="agent", category="GATE", action="constraint.profile.save", target={"type": "CONSTRAINT_PROFILE", "id": profile.get("profileId") or "default"}, power="P2", rationale={ "reason": str(exc), "constraintId": exc.constraint_id, "requested": payload, "result": "DENIED", }, result="DENIED", ) def _gate_hard_constraint_close(world: World, normalized: dict[str, Any]) -> None: """门禁：任何使硬约束失效的保存请求一律拒绝（中文说明）。 覆盖三种「关闭」路径，均先写 DENIED 审计再抛 ConstraintProfileDenied： 1) enabled=False 落在硬约束上（含同包 kind=hard + enabled=False） 2) 硬约束降级为软约束（规避发布门禁）；仅 kindConfigurable 类（C6/C13）可软硬互转 3) 对硬约束设置权重（软权重无法关闭硬约束） """ if normalized.get("resetDefaults"): return cur = {c["id"]: c for c in get_constraint_profile(world)["constraints"]} defaults = _index_defaults() for cid, patch in normalized["constraints"].items(): item = cur.get(cid) or defaults.get(cid) or {} name = item.get("name") or cid cur_kind = item.get("kind") or "hard" target_kind = patch.get("kind", cur_kind) kind_configurable = bool((defaults.get(cid) or {}).get("kindConfigurable")) if patch.get("enabled") is False and target_kind == "hard": raise ConstraintProfileDenied( f"{name}（{cid}）是硬约束，不可关闭：硬约束违反将阻止发布，" "关闭请求被门禁拒绝并已记录审计", constraint_id=cid, requested=patch) if patch.get("kind") == "soft" and cur_kind == "hard" and not kind_configurable: raise ConstraintProfileDenied( f"{name}（{cid}）是硬约束，不可降级为软约束以规避发布门禁；" "关闭请求被门禁拒绝并已记录审计", constraint_id=cid, requested=patch) if "weight" in patch and target_kind == "hard": raise ConstraintProfileDenied( f"{name}（{cid}）是硬约束，权重固定为 1；软权重无法关闭硬约束，" "请求被门禁拒绝并已记录审计", constraint_id=cid, requested=patch) def _normalize_or_denied(world: World, payload: dict[str, Any]) -> dict[str, Any]: """校验约束剖面载荷；关闭硬约束被门禁拒绝时写审计并抛 ConstraintProfileDenied。""" try: return normalize_profile_payload(payload, world=world) except ConstraintProfileDenied as exc: _write_profile_denied_audit(world, payload, exc) raise def normalize_profile_payload(payload: dict[str, Any], world: World | None = None) -> dict[str, Any]: """校验 constraint.profile.save 载荷。""" if payload.get("resetDefaults"): return {"resetDefaults": True} raw = payload.get("constraints") if not isinstance(raw, dict) or not raw: raise ValueError("constraints 必须是非空对象（constraintId → {enabled,kind,weight}）") defaults = _index_defaults() out: dict[str, dict[str, Any]] = {} for cid, patch in raw.items(): if cid not in defaults: raise ValueError(f"未知约束：{cid}") if not defaults[cid].get("configurable"): if patch.get("enabled") is False: raise ConstraintProfileDenied( f"{defaults[cid]['name']}（{cid}）为内建硬约束，不可在配置中心关闭：" "硬约束违反将阻止发布，关闭请求被门禁拒绝并已记录审计", constraint_id=cid, requested=patch) raise ValueError(f"{defaults[cid]['name']} 为内建硬约束，不可在配置中心修改") if not isinstance(patch, dict): raise ValueError(f"{cid} 配置必须是对象") entry: dict[str, Any] = {} if "enabled" in patch: entry["enabled"] = bool(patch["enabled"]) if "kind" in patch: if patch["kind"] not in ("hard", "soft"): raise ValueError(f"{cid}.kind 只能是 hard/soft") entry["kind"] = patch["kind"] if "weight" in patch: try: w = float(patch["weight"]) except (TypeError, ValueError) as exc: raise ValueError(f"{cid}.weight 必须是数字") from exc if w < 0 or w > 10: raise ValueError(f"{cid}.weight 须在 0~10") entry["weight"] = w if isinstance(patch.get("params"), dict): entry["params"] = dict(patch["params"]) if not entry: raise ValueError(f"{cid} 未提供可更新字段") out[cid] = entry if world is not None: _gate_hard_constraint_close(world, {"constraints": out}) return {"constraints": out} def confirmation_for_profile_save(world: World, payload: dict[str, Any]) -> tuple[str, list[str]]: norm = _normalize_or_denied(world, payload) if norm.get("resetDefaults"): return "恢复默认约束剖面", [ "物料齐套恢复为软约束并启用", "维保/产能/班组/模具等可配项恢复默认", "只影响后续新排产与发布门禁，不回写历史版本", ] cur = {c["id"]: c for c in get_constraint_profile(world)["constraints"]} lines: list[str] = [] for cid, patch in norm["constraints"].items(): name = cur[cid]["name"] bits = [] if "enabled" in patch: bits.append("启用" if patch["enabled"] else "关闭") if "kind" in patch: bits.append("硬约束" if patch["kind"] == "hard" else "软约束") if "weight" in patch: bits.append(f"权重={patch['weight']}") lines.append(f"{name}（{cid}）：" + " / ".join(bits)) lines.append("硬约束违反将阻止发布；软约束仅报告风险") lines.append("只影响后续新排产版本") return "更新约束配置", lines def apply_profile_save(world: World, payload: dict[str, Any]) -> dict[str, Any]: norm = _normalize_or_denied(world, payload) before = deepcopy(get_constraint_profile(world)) if norm.get("resetDefaults"): world["constraintProfile"] = { "profileId": "default", "name": "默认约束剖面", "constraints": {}, } else: profile = world.setdefault("constraintProfile", { "profileId": "default", "name": "默认约束剖面", "constraints": {}, }) stored = profile.setdefault("constraints", {}) for cid, patch in norm["constraints"].items(): prev = stored.get(cid) or {} merged = {**prev, **patch} if isinstance(prev.get("params"), dict) and isinstance(patch.get("params"), dict): merged["params"] = {**prev["params"], **patch["params"]} stored[cid] = merged after = get_constraint_profile(world) return { "kind": "CONSTRAINT_PROFILE", "id": after["profileId"], "name": after["name"], "before": before, "after": after, "reset": bool(norm.get("resetDefaults")), } def default_constraint_profile() -> dict[str, Any]: return {"profileId": "default", "name": "默认约束剖面", "constraints": {}}