# ============================================================ # 冲突解决工作流（moduleId: domain-conflicts, 可重生 ✅） # EX-03：冲突列表 + 可执行建议 + 一键应用（安全修复 P1 / 重排类 P2） # ============================================================ from __future__ import annotations from typing import Any World = dict[str, Any] # 冲突类型 → 默认修复动作（可被上下文改写） _FIX_CATALOG: dict[str, list[dict[str, Any]]] = { "NO_CAPABILITY": [ {"action": "flex.fault.recover", "label": "恢复设备后重排", "power": "P1"}, {"action": "flex.reschedule", "label": "L2 短窗重排", "power": "P2", "params": {"level": "L2"}}, {"action": "flex.swap", "label": "L1 局部换机（池内备机）", "power": "P1"}, ], "NO_MOLD": [ {"action": "flex.mold.unlock", "label": "解锁同工序模具", "power": "P1"}, {"action": "flex.reschedule", "label": "L2 短窗重排", "power": "P2", "params": {"level": "L2"}}, ], "MOLD_LIFE": [ {"action": "flex.mold.unlock", "label": "解锁并重置寿命计数", "power": "P1"}, {"action": "flex.reschedule", "label": "L2 短窗重排", "power": "P2", "params": {"level": "L2"}}, ], "DELAY": [ {"action": "flex.reschedule", "label": "L2 短窗重排", "power": "P2", "params": {"level": "L2"}}, {"action": "flex.reschedule", "label": "L3 日窗重排", "power": "P2", "params": {"level": "L3"}}, ], "WINDOW_DEFERRED": [ {"action": "flex.reschedule", "label": "L3 日窗重排（扩窗）", "power": "P2", "params": {"level": "L3"}}, {"action": "flex.schedule", "label": "中窗重排 mid", "power": "P1", "params": {"window": "mid"}}, ], "WINDOW_TRUNCATED": [ {"action": "flex.reschedule", "label": "L3 日窗重排", "power": "P2", "params": {"level": "L3"}}, ], "MATERIAL_SHORTAGE": [ {"action": "note", "label": "建议补料/调库存后重排（请走主数据）", "power": "P0"}, {"action": "flex.reschedule", "label": "L2 短窗重排（延期接受）", "power": "P2", "params": {"level": "L2"}}, ], "NO_TEAM": [ {"action": "note", "label": "建议增派班组人数后重排", "power": "P0"}, {"action": "flex.reschedule", "label": "L2 短窗重排", "power": "P2", "params": {"level": "L2"}}, ], "NO_ROUTING": [ {"action": "note", "label": "请补齐工艺路线主数据", "power": "P0"}, ], "EQUIPMENT": [ {"action": "schedule.run", "label": "固定轨试排（避让维保）", "power": "P1"}, ], "CAPACITY": [ {"action": "schedule.run", "label": "固定轨试排（产能均衡）", "power": "P1", "params": {"strategy": "CAPACITY_BALANCE"}}, ], "DELAY_FIXED": [ {"action": "schedule.run", "label": "交期优先试排", "power": "P1", "params": {"strategy": "DELIVERY_FIRST"}}, ], } def list_conflict_center(world: World, scope: str = "flex") -> dict: """冲突中心投影：最新版本未解决冲突 + 可执行建议。""" from server.state.seed import ensure_flex_seed ensure_flex_seed(world) rows: list[dict] = [] if scope in ("flex", "all"): versions = world.get("flexScheduleVersions", []) latest_id = versions[-1]["id"] if versions else None for cf in world.get("flexConflicts", []): if latest_id and cf.get("versionId") != latest_id: continue if cf.get("isResolved"): continue rows.append(_enrich(cf, track="flex")) if scope in ("fixed", "all"): versions = world.get("scheduleVersions", []) latest_id = versions[-1]["id"] if versions else None for cf in world.get("conflicts", []): if latest_id and cf.get("versionId") != latest_id: continue if cf.get("isResolved"): continue rows.append(_enrich(cf, track="fixed")) sev = {"CRITICAL": 0, "MAJOR": 1, "MINOR": 2} rows.sort(key=lambda r: (sev.get(r.get("severity") or "", 9), r.get("id") or 0)) return { "scope": scope, "total": len(rows), "critical": sum(1 for r in rows if r.get("severity") == "CRITICAL"), "major": sum(1 for r in rows if r.get("severity") == "MAJOR"), "rows": rows, } def _enrich(cf: dict, track: str) -> dict: ctype = cf.get("conflictType") or "" # 固定轨 DELAY 与柔性同名，用 catalog 键区分策略时可映射 catalog_key = ctype if track == "fixed" and ctype == "DELAY": catalog_key = "DELAY_FIXED" fixes = [] for raw in _FIX_CATALOG.get(catalog_key, []): fix = {**raw, "params": {**(raw.get("params") or {})}} # 注入冲突上下文 if fix["action"] == "flex.swap": # 从描述/资源名猜设备码 code = _guess_equipment(cf) if code: fix["params"]["equipmentCode"] = code else: continue # 无设备码则不展示换机 if fix["action"] == "flex.fault.recover": code = _guess_equipment(cf) if not code: continue fix["params"]["equipmentCode"] = code if fix["action"] == "flex.mold.unlock": mold = cf.get("resourceName") if cf.get("resourceType") == "MOLD" else None fix["params"]["moldCode"] = mold fix["params"]["operationHint"] = cf.get("operationCode") fixes.append(fix) if not fixes: fixes = [{"action": "note", "label": cf.get("suggestedSolution") or "人工处理", "power": "P0", "params": {}}] return { "id": cf.get("id"), "track": track, "conflictType": ctype, "severity": cf.get("severity"), "orderNo": cf.get("orderNo"), "resourceType": cf.get("resourceType"), "resourceName": cf.get("resourceName"), "description": cf.get("description"), "engineSuggestion": cf.get("suggestedSolution"), "versionId": cf.get("versionId"), "isResolved": bool(cf.get("isResolved")), "fixes": fixes, } def _guess_equipment(cf: dict) -> str | None: import re for src in (cf.get("resourceName"), cf.get("description"), cf.get("suggestedSolution")): if not src: continue m = re.search(r"(PRESS|WELD|CUT|TEST|LEAK|PULL|PKG)-\d+", str(src), re.I) if m: return m.group(0).upper() # 从工序码推断默认单点设备（演示种子） desc = str(cf.get("description") or "") m_op = re.search(r"OP-([A-Z]+)", desc, re.I) if m_op: op = m_op.group(1).upper() defaults = {"WELD": "WELD-01", "CRIMP": "PRESS-01", "CUT": "CUT-01", "TEST": "TEST-01", "LEAK": "LEAK-01"} return defaults.get(op) return None def apply_conflict_fix(store, conflict_id: int, action: str, params: dict | None = None, actor: str = "web", session_id: str = "web") -> dict: """应用冲突修复。P2 动作返回 stage 确认卡；P1 直接执行并标记冲突已解决。""" from server.agent_core import harness from server.agent_core.audit import write_audit from server.state.seed import ensure_flex_seed ensure_flex_seed(store.data) world = store.data params = dict(params or {}) cf = _find_conflict(world, conflict_id) if cf is None: raise ValueError(f"找不到冲突 id={conflict_id}") if cf.get("isResolved"): raise ValueError(f"冲突 #{conflict_id} 已解决") if action == "note": return {"message": "该建议需人工处理，未改世界状态。", "applied": False} if action == "flex.reschedule": from server.aps_domain.flex import preview_reschedule level = str(params.get("level") or "L2").upper() title, lines = preview_reschedule(store, level) lines = [f"针对冲突 #{conflict_id}：{cf.get('description', '')[:80]}", *lines] block = harness.stage_confirmation( session_id, "flex.reschedule", {"level": level, "sortMode": params.get("sortMode"), "resolveConflictId": conflict_id}, title=f"冲突修复 · {title}", summary_lines=lines) write_audit(world, store.next_id, actor=actor, category="GATE", action="conflict.apply.stage", target={"type": "FLEX_CONFLICT", "id": conflict_id}, power="P2", rationale={"confirmId": block.props["confirmId"], "action": action, "level": level}) store.save() return {"message": f"{title} 已进入 P2 确认队列。", "applied": False, "staged": True, "block": block.model_dump()} def _apply(): result: dict[str, Any] = {"action": action} if action == "flex.swap": from server.aps_domain.flex import local_swap_equipment code = params.get("equipmentCode") or _guess_equipment(cf) if not code: raise ValueError("无法定位故障设备，请手动局部换机") # local_swap 内部已 guard；此处再包一层仅记账 swap = local_swap_equipment(store, equipment_code=code, mark_fault=True, actor=actor) result["swap"] = swap elif action == "flex.fault.recover": from server.aps_domain.flex import apply_equipment_fault code = params.get("equipmentCode") or _guess_equipment(cf) if not code: raise ValueError("无法定位设备") result["fault"] = apply_equipment_fault( store, code, status="RUNNING", reschedule=True, actor=actor) elif action == "flex.mold.unlock": from server.aps_domain.flex import patch_flex_resource mold_code = params.get("moldCode") or ( cf.get("resourceName") if cf.get("resourceType") == "MOLD" else None) if not mold_code: # 找同工序已锁模具 mold_code = _find_locked_mold(world, params.get("operationHint")) if not mold_code: raise ValueError("找不到可解锁模具") mold = next(m for m in world["flexMolds"] if m["code"] == mold_code) life_total = int(mold.get("lifeTotal") or 0) used = int(mold.get("lifeUsed") or 0) patch = {"status": "AVAILABLE"} if life_total and used >= life_total: patch["lifeUsed"] = max(0, life_total - 1000) # 重置留一点余量演示 result["mold"] = patch_flex_resource(store, "mold", mold_code, patch, actor=actor) elif action == "flex.schedule": from server.aps_domain.flex import run_flex_schedule result["schedule"] = run_flex_schedule( store, sort_mode="BOTTLENECK", window=params.get("window") or "mid", actor=actor, name=f"冲突#{conflict_id}修复排产") elif action == "schedule.run": from server.engines import get_engine from server.engines.base import EngineParams from server.timeutil import add_minutes, fmt_date, today0 strategy = params.get("strategy") or "COMPREHENSIVE" ep = EngineParams( orderIds=[], engineType="RULE", strategyTemplate=strategy, planningHorizonDays=14, startDate=fmt_date(add_minutes(today0(), 24 * 60)), ) eng = get_engine("RULE") result["schedule"] = eng.solve(world, ep, store.next_id) store.save() else: raise ValueError(f"不支持的修复动作：{action}") cf["isResolved"] = True cf["resolutionAction"] = action cf["resolvedBy"] = actor return result power = harness.power_of( "flex.schedule" if action == "flex.schedule" else "flex.swap" if action == "flex.swap" else "flex.fault" if action == "flex.fault.recover" else "flex.resource.patch" if action == "flex.mold.unlock" else "schedule.run" if action == "schedule.run" else "flex.reschedule") if power == "P2": raise ValueError(f"动作 {action} 为 P2，请走确认卡路径") # 用 flex.conflict.resolve 统一门禁 out = harness.guard("flex.conflict.resolve", {"conflictId": conflict_id, "action": action}, _apply) write_audit(world, store.next_id, actor=actor, category="WORLD_WRITE", action="conflict.apply", target={"type": "CONFLICT", "id": conflict_id}, power="P1", rationale={"action": action, "params": params}) store.save() return {"message": f"冲突 #{conflict_id} 已应用「{action}」。", "applied": True, "result": out, "conflict": _enrich(cf, track="flex" if conflict_id in {c.get("id") for c in world.get("flexConflicts", [])} else "fixed")} def _find_conflict(world: World, conflict_id: int) -> dict | None: for cf in world.get("flexConflicts", []): if cf.get("id") == conflict_id: return cf for cf in world.get("conflicts", []): if cf.get("id") == conflict_id: return cf return None def _find_locked_mold(world: World, operation_hint: str | None) -> str | None: for m in world.get("flexMolds", []): if m.get("status") != "LOCKED": continue if operation_hint and m.get("operationCode") != operation_hint: continue return m["code"] for m in world.get("flexMolds", []): if m.get("status") == "LOCKED": return m["code"] return None