# ============================================================ # 排产工作流编排（moduleId: domain-workflow, 可重生 ✅） # 职责：意图 → 动作 → AgentReply（文本 + 视口命令 + UI 块） # 权力路由：P0/P1 直接执行；P2（发布/重置）经门禁出确认卡（§3.3） # ============================================================ from __future__ import annotations # 前向类型引用 import copy import hashlib import json import os import tempfile from pathlib import Path from typing import TYPE_CHECKING, Any # 类型标注 if TYPE_CHECKING: from server.agent_core.session_ref import RefResolutionError from server.agent_core import harness # 门禁 v1 from server.agent_core.audit import write_audit # 审计写入 from server.contracts import ( # 跨层契约 AgentReply, IntentResult, ScheduleResult, ViewportCommand, ) from server.engines import get_engine # 引擎工厂 from server.engines.base import EngineParams # 引擎入参 from server.state.checkpoints import get_checkpoints # 成对快照仓（§4.3） from server.state.store import WorldStore # 世界状态存储 from server.timeutil import add_minutes, fmt_date, today0 # 日期工具 from server.aps_domain.scenario import compare_scenarios # 方案对比（Explore 沙盒 §9.7） from server.aps_domain.reports import build_report, persist_report_xlsx # 报告生成 v1（§9.10） from server.knowledge import get_knowledge, get_preferences # 知识库与偏好仓（M3 §8） from server.knowledge.preferences import extract_features # 多特征规则归纳输入（矩阵 66） from server.contracts import UIBlock # 报告/证据块组装 # 帮助文案：能力清单（help 意图与拒识兜底共用） _HELP = "\n".join([ "可执行的排产能力：", "· 查询现场数据：有哪些订单 / 查订单 102285668 / 这个订单有委外吗 / 查物料 / 查 BOM / 查工艺路线", "· 知识库：机加工工艺怎么生成 / 车削工艺模式 / 装配工艺模式 / 知识清单 / 导入知识文档", "· 生成排产方案：交期优先试排 / 产能均衡重排 / 规则引擎试排", "· 柔性排产（能力池+虚拟产线）：生成柔性排产方案 / 按订单排产 / 正排 / 倒排 / 瓶颈锚", "· 外部算法：用外部算法试排 / 查看已接入算法 / skill 健康检查", "· 瓶颈产能：瓶颈产能 / 各工序能做多少（按能力池实时算）", "· 交期承诺：某产品 N 套什么时候能交", "· 方案评估：多策略方案对比（沙盒运行，不影响当前版本）", "· 紧急插单：插单快评 / 采用插单（需确认）", "· 预测订单：查看预测 / 新建预测 / 预测纳入试排", "· 切到负荷热力图 / 交期承诺看板 / 甘特图 / 柔性甘特", "· 标出超期订单 / 查看冲突 / 按小时看 / 重置视图", "· 建一个检查点 / 回滚到上一个检查点（需确认）", "· 换线有什么规定 / @知识:机加工工艺路线生成总则（知识库带出处）", "· 生成排产方案报告（下载 Excel 工作计划表） / 生成日报 / 生成版本对比报告", "· 查看当前KPI / 下一步建议 / 发布当前排产版本（需确认）/ 重置数据（需确认）", ]) # 策略模板的中文名（回复文案用） _STRATEGY_CN = {"DELIVERY_FIRST": "交期优先", "CAPACITY_BALANCE": "产能均衡", "COST_FIRST": "成本最优(换型代理)", "CHANGEOVER_MIN": "换型最小化", "CAMPAIGN": "战役合并", "FIFO": "先进先出", "COMPREHENSIVE": "综合优化", "KITTING_FIRST": "齐套优先", "SKILL_FIRST": "技能优先"} # OR-04：最近一次插单快评（供「采用插单」复用；进程内存） _LAST_RUSH_EVAL: dict[str, Any] = {} def _build_run_trace(result: ScheduleResult, params: EngineParams, intent: IntentResult, store: WorldStore) -> dict: """构造 schedule.run 可追溯链（尽力而为；失败返回空链摘要不影响求解）。 链哈希只包含确定性信息：版本/run/引擎/策略/求解状态/gap/输入包指纹。 墙钟 solveTimeSec 不进入链哈希（同输入同种子跨引擎可复算）， 但仍保留在 ALGO_RUN 审计 rationale 顶层字段 solveTimeSec 供人查看。 """ try: from server.agent_core.evidence import EvidenceItem, trace_chain, evidence_ref version_ref = evidence_ref("schedule-version", result.versionId) items = [ EvidenceItem(kind="schedule-version", ref=str(result.versionId), version=str(result.versionId)), EvidenceItem(kind="run", ref=f"run-{result.versionId}", runId=str(result.versionId), engine=params.engineType, meta={"strategy": params.strategyTemplate, "solveStatus": result.solveStatus, "gap": result.optimalityGap}), EvidenceItem(kind="algorithm", ref=params.engineType, version=params.engineType, inputsHash=_run_inputs_hash(store.data, params)), ] chain = trace_chain(items) return { "chainHash": chain["chainHash"], "count": chain["count"], "summary": [{"kind": it["kind"], "ref": it["ref"]} for it in chain["items"]], "items": chain["items"], "versionRef": version_ref, } except Exception: return {"chainHash": None, "count": 0, "summary": [], "versionRef": None} def _run_inputs_hash(world: dict, params: EngineParams) -> str | None: """输入包指纹（可复算）：订单池 + 参数 + 引擎的稳定投影。""" import hashlib, json as _json try: payload = { "orders": len(world.get("salesOrders") or []) + len(world.get("flexOrders") or []), "engine": params.engineType, "strategy": params.strategyTemplate, "horizon": params.planningHorizonDays, "constraints": sorted(params.constraints.items()) if params.constraints else [], } return hashlib.sha256( _json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() except Exception: return None def _run_audit_evidence(result: ScheduleResult, trace: dict) -> list[str]: """合并引擎 evidenceRefs 与 trace 的 schedule-version 引用（去重保序）。 对齐 round-15 计划假设 2：evidenceRefs 保留 schedule-version ref； trace 构造失败（versionRef=None）时仅保留引擎原始引用，不阻断审计。 """ refs = list(result.evidenceRefs or []) version_ref = trace.get("versionRef") if version_ref and version_ref not in refs: refs.append(version_ref) return refs def _run_schedule(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply: """执行试排（P1：写草稿版本 + 落盘；发布另走 P2 门禁）。""" # 偏好个性化（§8.3 第一层）：用户没点名策略时用其历史最常用策略 explicit = intent.params.get("strategy") # 用户明确点名的策略（可能为空） pref_note = "" # 个性化说明（透明可解释） if explicit: # 点名 → 直接用 strategy = explicit else: # 未点名 → 查偏好仓 strategy, scores = get_preferences().preferred_strategy( default="COMPREHENSIVE", project_id=getattr(store, "world_key", None)) if scores: # 有历史信号才标注（避免误导） pref_note = f"\n（按你的使用偏好选了{_STRATEGY_CN.get(strategy, strategy)}，历史使用 {scores.get(strategy, 0)} 次；点名策略可覆盖）" from server.aps_domain.params import get_schedule_params from server.aps_domain.constraints import engine_constraint_flags sp = get_schedule_params(store.data) params = EngineParams( # 组装引擎入参 orderIds=[], # M1：全部待排订单 engineType=intent.params.get("engine") or sp.get("defaultEngine") or "HYBRID", strategyTemplate=strategy, # 策略模板（点名或偏好） planningHorizonDays=int(sp.get("planningHorizonDays") or 14), # 展望期（OR-02 可配） startDate=fmt_date(add_minutes(today0(), 24 * 60)), # 明天开排 includeForecast=bool(intent.params.get("includeForecast")), # OR-05：预测纳入试排 constraints=engine_constraint_flags(store.data), # SC-04 约束剖面 → 引擎开关 timeLimitSeconds=float(sp["cpTimeLimitSeconds"]) if sp.get("cpTimeLimitSeconds") is not None else 8.0, ) engine = get_engine(params.engineType) # CP / HYBRID 真管线；GA 仍 RULE 代跑 result: ScheduleResult = engine.solve(store.data, params, store.next_id) # 求解（写内存世界） # 可追溯链：run-id / 算法版本 / 种子 / 知识版本 / 用户确认 串成一条链（§8.4） trace = _build_run_trace(result, params, intent, store) # 审计：算法运行留痕（§3.6.2 算法运行类：引擎/策略/计数可复算） write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="schedule.run", target={"type": "SCHEDULE_VERSION", "id": result.versionId}, power="P1", rationale={"strategy": params.strategyTemplate, "engine": params.engineType, "evidence": _run_audit_evidence(result, trace), "solveStatus": result.solveStatus, "solveTimeSec": result.solveTimeSec, "gap": result.optimalityGap, "traceChainHash": trace["chainHash"], "traceCount": trace["count"], "traceSummary": trace["summary"], "traceItems": trace.get("items")}) store.save() # 试排结果落盘（草稿版本也持久化） get_preferences().record(params.strategyTemplate, source="schedule.run", actor=actor, project_id=getattr(store, "world_key", None), features=extract_features(store.data)) # 偏好信号沉淀（§8.3 项目隔离 + 矩阵 66 特征规则输入） engine_note = "" if params.engineType in ("CP", "HYBRID"): st = result.solveStatus or "—" gap = result.optimalityGap gap_txt = f"{gap:.2%}" if gap is not None else "—" tsec = result.solveTimeSec t_txt = f"{tsec:.2f}s" if tsec is not None else "—" pipe = "RULE→CP-SAT" if params.engineType == "HYBRID" else "CP-SAT" engine_note = ( f"\n（{pipe}：状态 {st} · 耗时 {t_txt} · gap {gap_txt}；工序落槽=班次占槽）" ) elif params.engineType == "GA": engine_note = "\n（注：GA 真管线尚未接入，本次由规则引擎代跑并留 engineType 痕）" text = (f"排产完成 ✅ 版本 {result.versionNo}（{_STRATEGY_CN.get(result.strategy, result.strategy)}）\n" f"生产订单 {result.poCount} 个 / 工单 {result.woCount} 个" + ("（含预测）" if params.includeForecast else "") + "\n" f"冲突 {result.conflictCount} 项 · 总延迟 {round(result.totalTardiness)}h · " f"平均利用率 {round(result.avgUtilization * 100)}%{engine_note}{pref_note}\n" f"右侧甘特已刷新为该版本。执行“发布当前排产版本”可进入 P2 发布确认。") # 回复正文（含下一步引导） blocks: list[UIBlock] = [] if result.conflictCount >= 3: from server.aps_domain.guidance import build_guidance blocks.append(UIBlock( blockId="guidance-after-run", type="guidance", props=build_guidance(store.data, context="high_conflict"))) return AgentReply(text=text, blocks=blocks, commands=[ # 联动命令：切甘特 + 重拉世界数据 ViewportCommand(cmd="viewport.mode", params={"mode": "gantt"}, issuedBy="LLM"), ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM"), ]) def _publication_safety_block(store: WorldStore, version: dict[str, Any]) -> str | None: """Return a fail-closed reason when a version must not become an execution baseline.""" snapshot = version.get("inputSnapshot") snapshot_context = snapshot.get("planningContext") if isinstance(snapshot, dict) else None current_context = store.data.get("planningContext") trial_only = ( version.get("trialOnly") is True or (isinstance(snapshot_context, dict) and snapshot_context.get("trialOnly") is True) or (isinstance(current_context, dict) and current_context.get("trialOnly") is True) ) reasons: list[str] = [] if trial_only: reasons.append("版本标记为 trialOnly=true（试排或历史资料）") if version.get("productionReady") is False: reasons.append("版本标记为 productionReady=false") return "；".join(reasons) if reasons else None _PUBLISH_REASON_TITLES: dict[str, str] = { "TRIAL_ONLY": "当前版本为试排草稿", "PRODUCTION_NOT_READY": "生产条件未确认", "VERSION_NOT_DRAFT": "版本状态不允许发布", "VL_COUNT_MISMATCH": "产线数量不一致", "WO_COUNT_MISMATCH": "工单数量不一致", "ORDER_COVERAGE_MISMATCH": "订单覆盖不完整", "UNSCHEDULED_REQUIREMENTS": "仍有未排订单", "HARD_CONFLICT": "存在未解决硬约束冲突", "HARD_BLOCKER": "存在发布阻断项", "HARD_VALIDATION_VIOLATION": "硬约束校验未通过", "NO_WORK_ORDERS": "没有可执行工单", "NO_ROUTING": "缺少工艺路线", "INCOMPLETE_ROUTING": "工艺路线不完整", "INCOMPLETE_WORK_ORDER": "工单字段不完整", "ORPHAN_WORK_ORDER": "工单缺少产线归属", "ORPHAN_VIRTUAL_LINE": "产线没有工单", "VERSION_MARKED_NOT_PUBLISHABLE": "版本被标记为不可发布", "DUPLICATE_OR_MISSING_VL_ID": "产线编号重复或缺失", "DUPLICATE_OR_MISSING_WO_ID": "工单编号重复或缺失", "DUPLICATE_WORK_ORDER_ACTIVITY_ID": "工序编号重复", "V2_EVIDENCE_REQUIRED": "缺少排产过程证据", "INCOMPLETE_V2_EVIDENCE": "排产过程证据字段不完整", "INVALID_V2_EVIDENCE": "排产过程证据无效", "V2_SOLUTION_NOT_FEASIBLE": "求解结果不是可行解", "V2_VALIDATION_NOT_VALID": "求解结果未通过校验", "V2_VALIDATION_HAS_HARD_VIOLATIONS": "求解结果仍违反硬约束", "V2_PROBLEM_NOT_FOUND": "找不到该版本的排产问题记录", "V2_ACTIVITY_COVERAGE_MISMATCH": "工序未全覆盖", "V2_PROBLEM_ID_DRIFT": "排产问题已变化", "V2_PROBLEM_REVISION_DRIFT": "排产问题版本已变化", "V2_PROBLEM_SOURCE_DRIFT": "排产问题来源数据已变化", "V2_SOLUTION_PROBLEM_DRIFT": "求解结果对应的排产问题已变化", "V2_SOLUTION_SOURCE_DRIFT": "求解结果来源数据已变化", "V2_SOLUTION_STATUS_DRIFT": "求解状态已变化", "V2_VALIDATION_IDENTITY_DRIFT": "校验证据标识已变化", "V2_VALIDATION_PROBLEM_DRIFT": "校验证据对应的排产问题已变化", "CURRENT_RESOURCE_CONSTRAINT_DRIFT": "资源约束已变化", "WORK_ORDER_ACTIVITY_NOT_IN_PROBLEM": "工序不在排产问题中", "WORK_ORDER_ACTIVITY_NOT_IN_SOLUTION": "工序不在求解结果中", "WORK_ORDER_ACTIVITY_IDENTITY_DRIFT": "工序标识已变化", "WORK_ORDER_START_DRIFT": "工序开始时间已变化", "WORK_ORDER_END_DRIFT": "工序结束时间已变化", "WORK_ORDER_OPERATION_DRIFT": "工序内容已变化", "WORK_ORDER_REQUIREMENT_ID_DRIFT": "需求编号已变化", "WORK_ORDER_RESOURCE_ALLOCATION_DRIFT": "资源分配已变化", "CP_SOLVE_NOT_MATERIALIZABLE": "CP 求解结果无法落成工序计划", "CP_C3_MATERIALIZATION_INVALID": "日历约束落细校验未通过", "CP_C7_MATERIALIZATION_INVALID": "产能约束落细校验未通过", "CP_TIMING_NOT_VALIDATED": "CP 时序未通过校验", "CP_TIMING_NOT_DIRECTLY_MATERIALIZED": "CP 时序未直接落细", } def _publication_gate_summary( store: WorldStore, version: dict[str, Any], *, track: str, validation: dict[str, Any] | None = None, ) -> dict[str, Any]: """Build a human-readable publication gate for chat and result cards.""" snapshot = version.get("inputSnapshot") snapshot_context = snapshot.get("planningContext") if isinstance(snapshot, dict) else None current_context = store.data.get("planningContext") trial_only = ( version.get("trialOnly") is True or (isinstance(snapshot_context, dict) and snapshot_context.get("trialOnly") is True) or (isinstance(current_context, dict) and current_context.get("trialOnly") is True) ) status = str(version.get("status") or "").upper() reasons: list[dict[str, str]] = [] checks: list[str] = [] if trial_only: reasons.append({ "code": "TRIAL_ONLY", "title": _PUBLISH_REASON_TITLES["TRIAL_ONLY"], "detail": "版本标记 trialOnly=true；结果来自历史资料或试排假设，不能作为正式执行基准。", }) else: checks.append("试排限制：未启用") if version.get("productionReady") is False: reasons.append({ "code": "PRODUCTION_NOT_READY", "title": _PUBLISH_REASON_TITLES["PRODUCTION_NOT_READY"], "detail": "版本标记 productionReady=false；人员技能、在制状态或供料条件尚未全部确认。", }) elif version.get("productionReady") is True: checks.append("生产就绪：已确认") else: checks.append("生产就绪：未标记为阻断") if status == "DRAFT": checks.append("版本状态：草稿") else: reasons.append({ "code": "VERSION_NOT_DRAFT", "title": _PUBLISH_REASON_TITLES["VERSION_NOT_DRAFT"], "detail": f"当前版本状态为 {status or 'UNKNOWN'}；只有草稿版本可以进入发布审批。", }) if validation: for row in validation.get("publishBlockingReasons") or []: code = str(row.get("code") or "PUBLICATION_CHECK_FAILED") if code == "VERSION_NOT_DRAFT": continue reasons.append({ "code": code, "title": _PUBLISH_REASON_TITLES.get(code, "发布校验未通过"), "detail": str(row.get("message") or code), }) if validation.get("structuralValid"): checks.append("版本结构：校验通过") if validation.get("publishReady") is True: checks.append("发布校验：通过") deduped: list[dict[str, str]] = [] seen: set[tuple[str, str]] = set() for reason in reasons: key = (reason["code"], reason["detail"]) if key in seen: continue seen.add(key) deduped.append(reason) if validation and validation.get("publishReady") is not True and not deduped: deduped.append({ "code": "PUBLICATION_CHECK_FAILED", "title": "发布校验未通过", "detail": "发布校验未返回可识别原因，请检查版本结构与生产就绪状态。", }) can_publish = not deduped source_label = "历史资料试排" if trial_only else ("柔性排产版本" if track == "flex" else "固定排产版本") if can_publish: next_steps = [ "确认发布范围与版本号", "完成 P2 审批，建立执行基准", "如需下发车间，另走 P3 MES 双人确认", ] elif trial_only: next_steps = [ "核对并补齐试排假设涉及的人员技能、在制状态与供料信息", "处理硬约束冲突与交期超期", "使用确认后的数据重新生成正式排产版本", "重新提交 P2 发布审批", ] else: next_steps = [ "按未满足条件修正版本数据或冲突", "重新生成并校验排产版本", "再次提交 P2 发布审批", ] return { "canPublish": can_publish, "statusLabel": "可进入发布审批" if can_publish else "不可发布为执行基准", "summary": ( "该版本已通过当前发布校验，可以进入 P2 人工审批。" if can_publish else f"当前为{source_label}，仍有发布条件未满足。" ), "sourceLabel": source_label, "reasons": deduped, "checks": checks, "nextSteps": next_steps, "boundary": "发布只建立执行基准，不会自动下发 MES；MES 下发仍需独立 P3 双人确认。", } def _publication_gate_block( version: dict[str, Any], *, track: str, gate: dict[str, Any], ) -> UIBlock: """Render a publication gate as the existing structured text block.""" version_no = version.get("versionNo") or version.get("id") return UIBlock( blockId=f"publication-gate-{track}-{version.get('id')}", type="text", props={ "title": f"发布条件 · {version_no}", "publicationGate": gate, }, ) def stage_schedule_publish( store: WorldStore, *, session_id: str, actor: str, track: str = "fixed", version_id: int | None = None, ) -> AgentReply: """Stage a P2 publication card for one exact fixed or flex version.""" normalized_track = str(track or "fixed").lower() if normalized_track not in {"fixed", "flex"}: return AgentReply(text=f"未知排产轨道：{normalized_track}。") version_key = "flexScheduleVersions" if normalized_track == "flex" else "scheduleVersions" versions = store.data.get(version_key) or [] version = ( next((row for row in versions if row.get("id") == version_id), None) if version_id is not None else (versions[-1] if versions else None) ) if version is None: from server.aps_domain.guidance import build_guidance block = UIBlock( blockId="guidance-no-version", type="guidance", props=build_guidance(store.data, context="empty_schedule"), ) return AgentReply( text="当前没有可发布的排产版本。请先生成并校验排产草案。", blocks=[block], ) safety_block = _publication_safety_block(store, version) if safety_block: gate = _publication_gate_summary(store, version, track=normalized_track) return AgentReply( text=( f"版本 {version.get('versionNo') or version['id']} 不可发布：{safety_block}。" "试排或未达到生产条件的版本只能保留为草稿，禁止建立执行基准或进入 MES 下发。" ), blocks=[_publication_gate_block(version, track=normalized_track, gate=gate)], ) status = str(version.get("status") or "").upper() if status in {"PUBLISHED", "DISPATCHED"}: return AgentReply(text=f"版本 {version.get('versionNo') or version['id']} 已发布，无需重复发布。") if status != "DRAFT": gate = _publication_gate_summary(store, version, track=normalized_track) return AgentReply( text=f"版本 {version.get('versionNo') or version['id']} 状态为 {status or 'UNKNOWN'}，不可发布。", blocks=[_publication_gate_block(version, track=normalized_track, gate=gate)], ) evidence_refs = [f"schedule-version:{version['id']}"] from server.aps_domain.mes import validate_dispatchable_version validation = validate_dispatchable_version( store.data, normalized_track, int(version["id"]), ) gate = _publication_gate_summary( store, version, track=normalized_track, validation=validation, ) if not validation.get("publishReady"): reasons = validation.get("publishBlockingReasons") or [] detail = "；".join( str(reason.get("message") or reason.get("code")) for reason in reasons[:5] ) track_name = "柔性" if normalized_track == "flex" else "固定" return AgentReply( text=( f"{track_name}版本 {version.get('versionNo') or version['id']} " f"未通过发布校验：{detail or '结构不完整'}" ), blocks=[_publication_gate_block(version, track=normalized_track, gate=gate)], ) if validation.get("evidenceRef"): evidence_refs.append(str(validation["evidenceRef"])) if normalized_track == "flex": summary_lines = [ f"闭环制造需求 {int(version.get('demandCount') or version.get('orderCount') or 0)} 项", f"产线 {int(version.get('vlCount') or 0)} 条 / 工单 {int(version.get('woCount') or 0)} 个", "发布只建立执行基准，不触发 MES；MES 下发仍需独立 P3 双人确认", ] target_type = "FLEX_VERSION" else: summary_lines = [ f"生产订单 {int(version.get('poCount') or 0)} 个 / 工单 {int(version.get('woCount') or 0)} 个将转入执行准备", f"未解决冲突 {int(version.get('conflictCount') or 0)} 项 · 总延迟 {round(float(version.get('totalTardiness') or 0))}h", "发布后该版本成为执行基准（P2 写主干世界状态）", ] target_type = "SCHEDULE_VERSION" params = { "track": normalized_track, "versionId": int(version["id"]), "evidenceRefs": evidence_refs, } block = harness.stage_confirmation( session_id, "schedule.publish", params, title=f"发布排产版本 {version.get('versionNo') or version['id']}", summary_lines=summary_lines, evidence_refs=evidence_refs, ) write_audit( store.data, store.next_id, actor=actor, category="GATE", action="schedule.publish.stage", target={"type": target_type, "id": version["id"]}, power="P2", rationale={ "confirmId": block.props["confirmId"], "track": normalized_track, "validationEvidence": evidence_refs[-1], }, evidence_refs=evidence_refs, ) store.save() return AgentReply( text=f"发布版本 {version.get('versionNo') or version['id']} 属于 P2 写操作，需要确认后执行。", blocks=[_publication_gate_block(version, track=normalized_track, gate=gate), block], ) def _stage_publish(store: WorldStore, session_id: str, actor: str, *, prefer_flex: bool = False) -> AgentReply: """Stage publication, preferring the latest flexible draft for chat commands.""" flex_versions = store.data.get("flexScheduleVersions") or [] latest_flex = flex_versions[-1] if flex_versions else None use_flex = bool( prefer_flex and latest_flex and str(latest_flex.get("status") or "").upper() == "DRAFT" ) return stage_schedule_publish( store, session_id=session_id, actor=actor, track="flex" if use_flex else "fixed", version_id=int(latest_flex["id"]) if use_flex else None, ) def _stage_order_upsert(store: WorldStore, session_id: str, intent: IntentResult, actor: str) -> AgentReply: """对话侧新建/编辑订单（P2 → 确认卡）：解析槽位 + 成品，缺项则追问。""" from server.aps_domain.orders import (confirmation_for_order_action, find_order, find_product_by_hint) p = dict(intent.params or {}) order_no = str(p.get("orderNo") or "") existing = find_order(store.data, order_no=order_no) if order_no else None if order_no and existing is None: return AgentReply(text=f"没找到订单 {order_no}，无法修改。可在订单面板核对订单号。") payload: dict[str, Any] = {} if existing: # 编辑：以现单兜底未给字段 first = (existing.get("items") or [{}])[0] payload.update({ "id": existing["id"], "customerName": existing["customerName"], "customerLevel": existing["customerLevel"], "deliveryDate": existing["deliveryDate"], "priority": existing.get("priority", 5), "status": existing.get("status", "CONFIRMED"), "productId": first.get("productId"), "quantity": first.get("quantity", 1), "isRush": existing.get("isRush", False), "specialRequirements": existing.get("specialRequirements", ""), }) # 覆盖用户明确给出的字段 for k in ("customerName", "customerLevel", "deliveryDate", "priority", "quantity"): if p.get(k) not in (None, ""): payload[k] = p[k] if "isRush" in p: payload["isRush"] = p["isRush"] prod_hint = p.get("productCode") or p.get("productName") if prod_hint: prod = find_product_by_hint(store.data, str(prod_hint)) if prod is None: names = "、".join(m["name"] for m in store.data["materials"] if m["type"] == "FINISHED_PRODUCT") return AgentReply(text=f"没找到成品「{prod_hint}」。可选成品：{names}。") payload["productId"] = prod["id"] missing = [label for key, label in (("customerName", "客户名称"), ("productId", "产品"), ("deliveryDate", "交期"), ("quantity", "数量")) if not payload.get(key)] if missing: verb = "修改" if existing else "新建" return AgentReply(text=f"要{verb}订单还差：{'、'.join(missing)}。\n" "可以一次说清，例如：「新建订单 客户 康尼机电 产品 PDU配电单元 数量 500 交期 2026-08-20」。") try: title, lines = confirmation_for_order_action(store.data, "order.upsert", payload) except ValueError as exc: return AgentReply(text=f"订单信息有误：{exc}") block = harness.stage_confirmation(session_id, "order.upsert", payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="order.upsert.stage", target={"type": "SALES_ORDER", "id": payload.get("id") or "NEW"}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 订单写入，需要你确认。", blocks=[block]) def _stage_reset(store: WorldStore, session_id: str, actor: str) -> AgentReply: """把“重置数据”压入门禁（P2：破坏性操作必须确认）。""" block = harness.stage_confirmation( # 生成确认卡 session_id, "data.reset", {}, title="清空并重置演示数据", summary_lines=["将删除全部排产版本/工单/审计并重新播种（不可恢复）"]) return AgentReply(text="重置数据是破坏性操作（P2），需要你确认。", blocks=[block]) def _version_resolvable(world: dict, version_id: int) -> bool: """对象域版本一致性：versionId 指向的版本必须仍可解析（供统一证据校验）。 仅当世界确实带版本表（scheduleVersions/flexScheduleVersions）时执行解析； 世界根本没有版本表（测试用极简 store/空世界）视为不适用并放行， 由各动作自身逻辑（mes preview 重校验、publish 目标版本查找等）兜底。 """ has_tables = "scheduleVersions" in world or "flexScheduleVersions" in world if not has_tables: return True for key in ("scheduleVersions", "flexScheduleVersions"): for v in world.get(key) or []: if v.get("id") == version_id: return True return False def _conversation_side_snapshot() -> dict[str, Any] | None: """收集对话侧完整快照（project workspace + messages + Plan + 分支，尽力而为）。 矩阵 52 行成对快照验收：快照须包含完整对话、Plan、活动分支和世界。 任一部分不可用（如测试环境无认证身份）时返回 None，仅保存摘要，不阻断建档。 """ try: from server.state.projects import get_project_store from server.agent_core.plan_runtime import get_plan_store side: dict[str, Any] = { "workspace": get_project_store().snapshot(include_messages=True), } try: plan_store = get_plan_store() side["plans"] = { pid: list(versions) for pid, versions in plan_store._plans.items() } except Exception: side["plans"] = None try: from server.state.branches import get_branches store = getattr(get_project_store(), "data", None) session_id = None if isinstance(store, dict): session_id = store.get("activeSessionId") if session_id: side["branches"] = get_branches().tree(session_id) else: side["branches"] = None except Exception: side["branches"] = None return side except Exception: return None def _restore_conversation_side(side: dict[str, Any] | None) -> None: """回滚时成对恢复对话侧快照（project workspace + messages；尽力而为）。 对话侧快照不可用（None / 缺 workspace）时静默跳过——世界侧回滚仍有效； 恢复失败不阻断回滚主流程（世界已恢复，对话侧缺失由时间线/会话树补）。 """ if not isinstance(side, dict): return workspace = side.get("workspace") if not isinstance(workspace, dict): return try: from server.state.projects import get_project_store get_project_store().replace_workspace(workspace) except Exception: pass def _create_checkpoint(store: WorldStore, *, label: str, reason: str, conversation_note: str = "") -> dict[str, Any] | None: """统一建档入口：世界侧快照 + 对话侧完整快照成对入库（矩阵 52 行）。 建档后把新 checkpointId 锚定到当前活动分支（分支↔快照联动）： 分支切换时可据此成对恢复该时刻的世界+对话侧。 """ meta = get_checkpoints().create( store.data, label=label, reason=reason, conversation_note=conversation_note, conversation_side=_conversation_side_snapshot(), ) if meta: try: from server.state.branches import get_branches bs = get_branches(getattr(store, "tenant_uuid", "platform"), getattr(store, "world_key", "default")) session_id = _active_session_id() if session_id: bs.ensure_session(session_id) active_id = bs.active(session_id) branch = bs._branch(session_id, active_id) branch["checkpointId"] = meta["pairId"] bs._write() except Exception: pass return meta class _FolderCandidateStore: """Process-local candidate world; save is a no-op until the P2 transaction commits.""" def __init__(self, source: Any) -> None: self.data = copy.deepcopy(source.data) self.tenant_uuid = getattr(source, "tenant_uuid", "platform") self.world_key = getattr(source, "world_key", "default") self.path = getattr(source, "path", "memory://folder-candidate") self._counters = copy.deepcopy(getattr(source, "_counters", {})) self._fallback_max = max( ( int(row.get("id")) for rows in self.data.values() if isinstance(rows, list) for row in rows if isinstance(row, dict) and isinstance(row.get("id"), int) ), default=0, ) def next_id(self, kind: str) -> int: current = max(int(self._counters.get(kind) or 0), self._fallback_max) self._counters[kind] = current + 1 self._fallback_max = max(self._fallback_max, self._counters[kind]) return self._counters[kind] def save(self) -> None: return None def _active_session_id() -> str | None: """当前活动会话 id（尽力而为；无认证上下文时返回 None）。""" try: from server.state.projects import get_project_store snapshot = get_project_store().snapshot(include_messages=False) return snapshot.get("activeSessionId") except Exception: return None # --------------------------------------------------------------------------- # FB-03：受控配置文件写（policy.update / ops.config.apply 共用原语） # tmp + os.replace 原子写（失败不留下半个文件）+ 旧文件 .bak 备份（保留最近 5 份） # --------------------------------------------------------------------------- def _atomic_write_json(path, doc: Any) -> None: """原子写 JSON 文件：先写同目录 tmp，再 os.replace 提升（无中间态）。""" import json as _json import uuid as _uuid from pathlib import Path as _Path path = _Path(path) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(f"{path.name}.tmp-{_uuid.uuid4().hex[:8]}") tmp.write_text(_json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8") os.replace(tmp, path) def _backup_config_file(path, *, keep: int = 5) -> str | None: """旧文件备份为 .bak--，保留最近 keep 份（供人工回滚）。""" import shutil as _shutil import time as _time import uuid as _uuid from pathlib import Path as _Path path = _Path(path) if not path.is_file(): return None backup = path.with_name( f"{path.name}.bak-{_time.strftime('%Y%m%d-%H%M%S')}-{_uuid.uuid4().hex[:6]}") _shutil.copy2(path, backup) backups = sorted(path.parent.glob(f"{path.name}.bak-*"), key=lambda p: p.stat().st_mtime, reverse=True) for old in backups[keep:]: try: old.unlink() except OSError: # 清理失败不阻断主流程 pass return str(backup) def _sql_payload_row_count(payload: dict[str, Any] | None) -> int: """SQL 冻结载荷的业务行数：按真实列表长度计，不信任 stats（两者可能不一致）。""" total = 0 for key in ("orders", "materials", "equipment", "routing", "bom"): value = (payload or {}).get(key) if isinstance(value, list): total += len(value) return total def execute_confirmed(store: WorldStore, confirm_id: str, approve: bool, actor: str, note: str | None = None, *, governance_service: Any | None = None) -> str: """Execute or reject a staged P2/P3 action through the sole confirmation path. P3 actions execute only after the trusted decision record carries the final grant. note：审批意见（可选），随批准/驳回记录进审批历史。 Returns: 面向用户的结果文案。 """ target_store = harness.resolve_confirmation_store(confirm_id, store) try: pending = harness.take_confirmation(confirm_id, approve=approve, note=note) except PermissionError as exc: try: from server.agent_core.plan_orchestration import decide_plan_node decide_plan_node( confirm_id=confirm_id, approve=False, note=f"approval envelope denied: {exc}", actor=actor, ) except Exception: pass write_audit( store.data, store.next_id, actor=actor, category="GATE", action="approval.envelope.denied", target={"type": "CONFIRMATION", "id": confirm_id}, power="P2", rationale={"confirmId": confirm_id, "reason": str(exc)}, result="DENIED", ) store.save() return f"确认信封证据校验未通过，未执行任何变更：{exc}" if pending is None: # 过期/重复点击 return "该确认卡已失效（可能已处理过）。" if target_store is not None: store = target_store action, params = pending["action"], pending["params"] # 解构动作 evidence_refs = list(pending.get("evidenceRefs") or params.get("evidenceRefs") or []) staged_snapshot = pending.get("beforeSnapshot") cp = {"pairId": staged_snapshot} if staged_snapshot else None # GATE audit uses staged snapshot folder_envelope = ( params.get("boundAction") == "folder.schedule" or "folderPayloadDigest" in params or any(str(ref).startswith("folder-") for ref in evidence_refs) ) if folder_envelope and action != "folder.schedule": write_audit( store.data, store.next_id, actor=actor, category="GATE", action="folder.schedule.envelope.denied", target={"type": "FOLDER_PACK", "id": params.get("projectId") or "project"}, power="P2", rationale={"confirmId": confirm_id, "reason": "ACTION_BINDING_MISMATCH"}, result="DENIED", before_snapshot=str(staged_snapshot) if staged_snapshot else None, evidence_refs=evidence_refs, ) store.save() return "工程目录确认信封已损坏，未执行任何变更。" if pending.get("approvalDenied") and pending.get("separationRequired"): write_audit( store.data, store.next_id, actor=actor, category="GATE", action=action + ".confirm.denied", target=params, power="P3", rationale={ "confirmId": confirm_id, "reason": "SEPARATION_OF_DUTIES", "approvalStep": pending.get("approvalStep"), }, result="DENIED", before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return "P3 二次审批必须由另一名用户完成；当前批准未执行，确认卡仍有效。" if approve and pending.get("needsSecondConfirm"): write_audit(store.data, store.next_id, actor=actor, category="GATE", action=action + ".confirm.first", target=params, power="P3", rationale={"confirmId": confirm_id, "approvalStep": 1, "requiredApprovals": pending.get("requiredApprovals"), "approvers": pending.get("approvals")}, result="PENDING_SECOND_CONFIRM", before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return "第一重确认已记录。该 P3 外部副作用仍未执行，请再次批准完成二次确认。" if not approve: # ---- 驳回分支 ---- # Agent 自动编排（矩阵 110 行）：驳回 → L2 Plan 节点 FAILED（尽力而为） try: from server.agent_core.plan_orchestration import decide_plan_node decide_plan_node(confirm_id=confirm_id, approve=False, note=note, actor=actor) except Exception: pass write_audit(store.data, store.next_id, actor=actor, category="GATE", action=action + ".reject", target=params, power=harness.power_of(action), rationale={"confirmId": confirm_id}, result="DENIED", before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) # 驳回留痕 store.save() # 落盘 return "已驳回，未做任何变更。" # ---- 统一证据强校验（§3.4 证据链 v1）：fail closed，缺项/漂移/版本失效拒绝 ---- try: harness.verify_pending_evidence( pending, checkpoint_store=get_checkpoints(), version_lookup=lambda vid: _version_resolvable(store.data, vid), current_world=store.data, ) except PermissionError as exc: write_audit(store.data, store.next_id, actor=actor, category="GATE", action=action + ".evidence.denied", target=params, power=harness.power_of(action), rationale={"confirmId": confirm_id, "reason": str(exc)}, result="DENIED", before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return f"证据校验未通过，未执行任何变更：{exc}" if action == "folder.schedule": from server.aps_domain.folder_pack import verify_folder_schedule_binding try: binding = verify_folder_schedule_binding(params, store.data) expected_refs = { f"folder-source:{binding['sourceManifestDigest']}", f"folder-world:{binding['folderWorldFingerprint']}", f"folder-payload:{binding['folderPayloadDigest']}", f"folder-snapshot:{binding['beforeSnapshot']}", } if not staged_snapshot: raise PermissionError("工程目录确认缺少审批前快照，请重新发起") if binding["beforeSnapshot"] != str(staged_snapshot): raise PermissionError("工程目录确认的审批前快照不一致，请重新发起") if str(params.get("projectId")) != str(pending.get("projectId")): raise PermissionError("工程目录确认的项目作用域不一致，请重新发起") if str(params.get("sessionId")) != str(pending.get("sessionId")): raise PermissionError("工程目录确认的会话作用域不一致，请重新发起") if str(params.get("targetWorldKey")) != str(pending.get("worldKey")): raise PermissionError("工程目录确认的世界作用域不一致，请重新发起") if not expected_refs.issubset(set(evidence_refs)): raise PermissionError("工程目录确认缺少完整证据引用，请重新发起") except (OSError, PermissionError, ValueError) as exc: try: from server.agent_core.plan_orchestration import decide_plan_node decide_plan_node( confirm_id=confirm_id, approve=False, note=f"folder evidence denied: {exc}", actor=actor, ) except Exception: pass write_audit( store.data, store.next_id, actor=actor, category="GATE", action="folder.schedule.evidence.denied", target={"type": "FOLDER_PACK", "id": params.get("projectId") or "project"}, power="P2", rationale={"confirmId": confirm_id, "reason": str(exc)}, result="DENIED", before_snapshot=str(staged_snapshot) if staged_snapshot else None, evidence_refs=evidence_refs, ) store.save() return f"工程目录证据校验未通过，未执行任何变更：{exc}" # folder.schedule is approved only after its candidate world commits. if action != "folder.schedule": try: from server.agent_core.plan_orchestration import decide_plan_node decide_plan_node(confirm_id=confirm_id, approve=True, note=note, actor=actor) except Exception: pass if action == "agent.fallback.execute": # ---- 批准：智能兜底计划执行（FB-02）---- # 计划锁 + checkpoint 成对快照 + diff 验证（P2-DESIGN §3）： # 分支体全异常归并显式失败（§9.2 降级方案——checkpoint 前置保证最坏情况 # 退化为「一次显式失败且已回滚的确认」，绝无半写入不声明）。 from server.agent_core import fallback_lane try: fb_result = fallback_lane.execute_plan(store, pending, actor=actor, evidence_refs=evidence_refs) except Exception as exc: # noqa: BLE001 - 热路径归并显式失败（绝不抛出） fb_result = fallback_lane.ExecuteResult( run_id=str(params.get("runId") or ""), ok=False, status="failed", plan_fingerprint=str(params.get("planFingerprint") or ""), error_message=f"{type(exc).__name__}: {exc}", message=f"兜底执行出现编排器内部错误（{type(exc).__name__}: {exc}），" "未执行任何变更。") write_audit( # 成败都写（失败总账在 restore 之后补写） store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="agent.fallback.execute", target={"type": "FALLBACK_RUN", "id": fb_result.run_id}, power="P2", rationale={ "confirmId": confirm_id, "approver": actor, "runId": fb_result.run_id, "planFingerprint": fb_result.plan_fingerprint, "stepsExecuted": fb_result.steps_executed, "status": fb_result.status, **({"deviation": fb_result.deviation} if fb_result.deviation else {}), **({"reason": fb_result.error_message} if fb_result.status in ("denied", "failed") else {}), "rolledBack": fb_result.rolled_back, "rollbackVerified": fb_result.rollback_verified, "cpAfter": fb_result.cp_after or None, "verifyReport": fb_result.report_path, "executionLog": fb_result.execution_log, }, result=("SUCCESS" if fb_result.ok else "DENIED" if fb_result.status == "denied" else "FAILED"), before_snapshot=fb_result.cp_before or None, evidence_refs=evidence_refs) store.save() # 落盘 return fb_result.message if action == "agent.fallback.policy.update": # ---- 批准：P3 白名单治理（FB-03）---- # 原子写白名单文件（P3-DESIGN §2.4）：beforeSha256 漂移比对 → 冻结文档 # 复验 → tmp+replace 原子替换 + .bak 备份 → WORLD_WRITE 审计。 # 分支体全异常归并显式失败（热路径纪律，绝不抛出）。 from server.agent_core import fallback_highrisk wl_path = "" before_sha = after_sha = None result_status = "FAILED" try: wl_path = str(fallback_highrisk.default_whitelist_path()) before_sha = fallback_highrisk.file_sha256(wl_path) if before_sha != params.get("beforeSha256"): result_status = "DENIED" message = ("白名单治理被拒绝：出卡后白名单文件已变化（漂移比对不符），" "未执行任何变更，请重新发起。") else: new_doc = params.get("document") doc_err = fallback_highrisk.validate_whitelist_doc(new_doc) if doc_err or fallback_highrisk.canonical_sha256(new_doc) \ != str(params.get("documentSha256") or ""): result_status = "DENIED" message = (f"白名单治理被拒绝：冻结文档复验失败" f"（{doc_err or '文档指纹与出卡值不符'}），未执行任何变更。") else: backup = _backup_config_file(wl_path) _atomic_write_json(wl_path, new_doc) readback_err = fallback_highrisk.validate_whitelist_doc( json.loads(Path(wl_path).read_text(encoding="utf-8"))) if readback_err: raise RuntimeError(f"回读校验失败: {readback_err}") after_sha = fallback_highrisk.file_sha256(wl_path) result_status = "SUCCESS" message = (f"P3 白名单已更新 ✅（原子替换完成" f"{f'，旧文件已备份 {backup}' if backup else ''}）。" "下一次裁决即生效。") except Exception as exc: # noqa: BLE001 - 热路径归并显式失败（原子写保证无中间态） message = (f"白名单治理执行失败（{type(exc).__name__}: {exc}），" "未留下半个文件（原子写语义）。") write_audit( store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="agent.fallback.policy.update", target={"type": "FALLBACK_POLICY", "id": wl_path}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "beforeSha256": before_sha, "afterSha256": after_sha, "diff": params.get("diff")}, result=result_status, before_snapshot=str(staged_snapshot) if staged_snapshot else None, evidence_refs=evidence_refs) store.save() return message if action == "agent.fallback.ops.config.apply": # ---- 批准：S7 受控配置变更（FB-03）---- # P3 双人链终点（P3-DESIGN §7.3）：消费一次性 executionGrant → # beforeSha256 漂移比对 → tmp+replace 原子替换 + .bak 备份 → WORLD_WRITE 审计。 from server.agent_core import fallback_highrisk from server.agent_core.feature_flags import default_features_path target_path = "" before_sha = after_sha = None grant_id = str(pending.get("executionGrant") or "") result_status = "FAILED" try: if not harness.consume_execution_grant( grant_id, confirm_id=confirm_id, action=action, params=params): result_status = "DENIED" message = "配置变更被拒绝：执行授权无效或已被消费，未执行任何变更。" else: target_path = str(default_features_path()) param_err = fallback_highrisk.validate_config_apply_params(params) before_sha = fallback_highrisk.file_sha256(target_path) if param_err: result_status = "DENIED" message = f"配置变更被拒绝：{param_err}，未执行任何变更。" elif before_sha != params.get("beforeSha256"): result_status = "DENIED" message = ("配置变更被拒绝：出卡后配置文件已变化（漂移比对不符），" "未执行任何变更，请重新发起。") else: backup = _backup_config_file(target_path) _atomic_write_json(target_path, params["content"]) readback_err = fallback_highrisk.validate_features_config_doc( json.loads(Path(target_path).read_text(encoding="utf-8"))) if readback_err: raise RuntimeError(f"回读校验失败: {readback_err}") after_sha = fallback_highrisk.file_sha256(target_path) result_status = "SUCCESS" message = (f"配置已原子替换 ✅（{params.get('file')}" f"{f'，旧文件已备份 {backup}' if backup else ''}）。" "features.json 每次现读，替换即生效。") except Exception as exc: # noqa: BLE001 - 热路径归并显式失败（原子写保证无中间态） message = (f"配置变更执行失败（{type(exc).__name__}: {exc}），" "未留下半个文件（原子写语义）。") write_audit( store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="agent.fallback.ops.config.apply", target={"type": "FALLBACK_OPS_CONFIG", "id": params.get("file")}, power="P3", rationale={"confirmId": confirm_id, "approver": actor, "grantId": grant_id, "approvals": pending.get("approvals"), "beforeSha256": before_sha, "afterSha256": after_sha}, result=result_status, before_snapshot=str(staged_snapshot) if staged_snapshot else None, evidence_refs=evidence_refs) store.save() return message if action in { "governance.rule.create", "governance.rule.update", "governance.rule.delete", "governance.rule.enable", "governance.rule.disable", }: from server.aps_domain.governance_rules import GovernanceRuleService from server.auth.context import get_identity identity = get_identity(required=True) authenticated_actor = identity.username or str(identity.user_id) cp = _create_checkpoint( store, label="治理规则变更前基线", reason=f"auto:{action}", conversation_note=f"批准执行 {action}", ) service = governance_service or GovernanceRuleService(store_provider=lambda: store) try: result = service.apply_confirmed_change( action, params, actor=authenticated_actor, confirm_id=confirm_id, before_snapshot=str(cp["pairId"]), evidence_refs=evidence_refs, ) except Exception: try: get_checkpoints().delete(str(cp["pairId"])) except (KeyError, OSError, TypeError, ValueError): pass raise if action == "governance.rule.delete": return f"治理规则 {result['ruleId']} 已删除 ✅（变更前已自动建档）。" return f"治理规则 {result['ruleId']} 已更新 ✅（变更前已自动建档）。" if action == "schedule.publish": # ---- approve publication ---- track = str(params.get("track") or "fixed").lower() version_key = "flexScheduleVersions" if track == "flex" else "scheduleVersions" version = next( (row for row in store.data.get(version_key) or [] if row.get("id") == params.get("versionId")), None, ) if version is None: return "发布被拒绝：确认卡绑定的排产版本已不存在。" safety_block = _publication_safety_block(store, version) if safety_block: return f"发布被拒绝：{safety_block}。版本保持草稿状态，未建立执行基准，也未下发 MES。" version_status = str(version.get("status") or "").upper() if version_status in {"PUBLISHED", "DISPATCHED"}: cp = _create_checkpoint( store, label=f"发布幂等前基线 {version.get('versionNo') or version['id']}", reason="auto:publish-idempotent", conversation_note=f"重复批准发布 {version.get('versionNo') or version['id']}", ) write_audit( store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="schedule.publish", target={ "type": "FLEX_VERSION" if track == "flex" else "SCHEDULE_VERSION", "id": version["id"], }, power="P2", rationale={ "confirmId": confirm_id, "approver": actor, "track": track, "idempotent": True, "existingStatus": version_status, }, result="SUCCESS", before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs, ) store.save() return f"版本 {version.get('versionNo') or version['id']} 已发布，本次批准按幂等执行并留痕。" if version_status != "DRAFT": return f"发布被拒绝：版本 {version.get('versionNo') or version['id']} 已不再是 DRAFT。" staged_evidence = [ ref for ref in evidence_refs if str(ref).startswith("schedule-evidence:") ] current_evidence_ref = None if track == "flex" or staged_evidence: from server.aps_domain.mes import validate_dispatchable_version validation = validate_dispatchable_version(store.data, track, int(version["id"])) if not validation.get("publishReady"): reasons = validation.get("publishBlockingReasons") or [] detail = "\uFF1B".join( str(reason.get("message") or reason.get("code")) for reason in reasons[:5] ) track_name = "柔性" if track == "flex" else "固定" return f"{track_name}版本发布被结构校验拦截：{detail or '版本不可发布'}" current_evidence_ref = validation.get("evidenceRef") if staged_evidence and current_evidence_ref not in staged_evidence: track_name = "柔性" if track == "flex" else "固定" return f"{track_name}版本发布被拒绝：出卡后排产结构或证据已漂移。" else: from server.aps_domain.constraints import hard_blocking_conflicts blockers = hard_blocking_conflicts( store.data, version_id=version["id"], track="fixed", ) if blockers: return ( f"发布被硬约束门禁拦截：仍有 {len(blockers)} 项未解决硬约束冲突。" "请先处理冲突或调整约束配置。" ) cp = _create_checkpoint( store, label=f"发布前基线 {version.get('versionNo') or version['id']}", reason="auto:publish", conversation_note=f"批准发布 {version.get('versionNo') or version['id']}", ) version["status"] = "PUBLISHED" from datetime import datetime from server.timeutil import fmt_dt version["publishedAt"] = fmt_dt(datetime.now()) if track == "fixed": for production_order in store.data.get("productionOrders") or []: if production_order.get("schedulingVersionId") == version["id"]: production_order["status"] = "CONFIRMED" write_audit( store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="schedule.publish", target={ "type": "FLEX_VERSION" if track == "flex" else "SCHEDULE_VERSION", "id": version["id"], }, power="P2", rationale={ "confirmId": confirm_id, "approver": actor, "track": track, "validationEvidence": current_evidence_ref, }, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs, ) store.save() suffix = "MES 下发仍需独立 P3 双人确认。" if track == "flex" else "生产订单已转入执行准备。" return f"版本 {version.get('versionNo') or version['id']} 已发布 ✅ {suffix}" if action == "data.reset": # ---- 批准：重置数据 ---- # 回滚防线：重置前自动建档（重置本身也可被撤销） cp = _create_checkpoint(store, label="重置前基线", reason="auto:reset", conversation_note="批准重置数据") store.reset() # 重新播种（含落盘） write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="data.reset", target={"type": "WORLD", "id": 0}, power="P2", rationale={"confirmId": confirm_id, "approver": actor}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) # 重置留痕 store.save() # 审计落盘 return "数据已重置为种子状态 ✅（重置前状态已自动存档，可回滚）" if action == "checkpoint.rollback": # ---- 批准：回滚到检查点 ---- pair = get_checkpoints().get(params["pairId"]) # 取目标快照（完整世界） if pair is None: # 快照不存在（被淘汰） return "目标检查点不存在（可能已被容量策略淘汰）。" # 回滚防线：回滚前先把"现在"也存档（允许撤销这次回滚——时间旅行可往返 §4.4） cp = _create_checkpoint(store, label="回滚前状态", reason="auto:rollback", conversation_note=f"回滚到 {pair['label']}") store.restore(pair["world"]) # 整体替换主干世界（成对恢复的世界侧） _restore_conversation_side(pair.get("conversationSide")) # 成对恢复对话侧（矩阵 52 行） write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="checkpoint.rollback", target={"type": "CHECKPOINT", "id": params["pairId"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "label": pair["label"]}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) # 回滚留痕 store.save() # 落盘 return f"已回滚到检查点【{pair['label']}】（{pair['createdAt']}）✅ 回滚前状态已自动存档。" if action in ("order.upsert", "order.cancel", "order.complete", "order.delete", "order.clear", "order.approve", "order.reject"): # ---- 批准：订单写入 / 审核 ---- from server.aps_domain.orders import apply_order_action cp = _create_checkpoint(store, label="订单变更前基线", reason=f"auto:{action}", conversation_note=f"批准执行 {action}") # P2 写前自动建档 applied = apply_order_action(store.data, store.next_id, action, params) order = applied["order"] write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action=action, target={"type": "SALES_ORDER", "id": order["id"], "orderNo": order["orderNo"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "beforeStatus": applied.get("beforeStatus"), "afterStatus": order["status"], "approvedCount": applied.get("approvedCount"), "rejectedCount": applied.get("rejectedCount")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() if action == "order.approve": return (f"已批准 ✅ {applied.get('approvedCount', 1)} 条订单 → APPROVED，" f"可参与正式排产：{'、'.join(applied.get('orderNos') or [order['orderNo']])}") if action == "order.reject": return (f"已驳回 ✅ {applied.get('rejectedCount', 1)} 条：" f"{'、'.join(applied.get('orderNos') or [order['orderNo']])}") if action == "order.upsert": verb = "已新增" if applied.get("created") else "已更新" return f"{verb}订单 {order['orderNo']} ✅ 后续试排会使用最新订单池。" if action == "order.cancel": return f"订单 {order['orderNo']} 已取消 ✅ 后续试排将不再纳入。" if action == "order.delete": return f"订单 {order['orderNo']} 已删除 ✅ 已从订单池移除。" if action == "order.clear": return (f"订单池已清空 ✅ 销售 {applied.get('clearedOrders', 0)} 条 / " f"采购 {applied.get('clearedPurchase', 0)} / 委外 {applied.get('clearedOutsource', 0)}。" "后续可重新导入或新建订单。") return f"订单 {order['orderNo']} 已完成 ✅ 后续试排将不再纳入。" if action == "rush.apply": # ---- OR-04：采用紧急插单（LNS 局部 / 升级全量）---- from server.aps_domain.rush import apply_rush lns = params.get("lns") or {} cp = _create_checkpoint(store, label="插单采用前基线", reason="auto:rush.apply", conversation_note="采用紧急插单前自动建档") if lns.get("status") == "LOCAL": # LNS 成功 → 局部方案入 DRAFT 草稿（窗口内最小扰动） from server.aps_domain.lns import apply_lns_local applied = apply_lns_local(store.data, store.next_id, params) order = applied["order"] result = applied["result"] d = applied["disturbance"] or {} write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="rush.apply", target={"type": "SALES_ORDER", "id": order["id"], "orderNo": order["orderNo"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "versionNo": result.versionNo, "strategy": applied.get("strategy"), "evalId": params.get("evalId"), "mode": "LNS_LOCAL", "lnsId": lns.get("lnsId"), "movedOrderCount": d.get("movedOrderCount"), "shiftedHours": d.get("shiftedHours"), "tardinessDelta": d.get("tardinessDelta")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() _LAST_RUSH_EVAL.clear() return (f"插单方案已采用 ✅ {order['orderNo']}（急单）已加入订单池；" f"{d.get('movedOrderCount', 0)} 张订单顺延，合计顺延 {d.get('shiftedHours', 0)} 小时，" f"总延期变化 {d.get('tardinessDelta', 0):+.1f} 小时。" f"已生成待发布的调整方案 {result.versionNo}，可用检查点回滚到采用前。") applied = apply_rush(store.data, store.next_id, params) # 升级（ESCALATE）/ 直接采用 → 全量重排 order = applied["order"] result = applied["result"] write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="rush.apply", target={"type": "SALES_ORDER", "id": order["id"], "orderNo": order["orderNo"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "versionNo": result.versionNo, "strategy": applied.get("strategy"), "evalId": params.get("evalId"), "mode": "FULL_RESCHEDULE" if lns.get("status") == "ESCALATE" else "DIRECT", "escalateReasons": lns.get("escalateReasons")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() _LAST_RUSH_EVAL.clear() prefix = "整版调整方案已采用 ✅ " if lns.get("status") == "ESCALATE" else "插单方案已采用 ✅ " return (prefix + f"{order['orderNo']}（急单）已加入订单池；" f"已生成待发布的调整方案 {result.versionNo}，共安排 {result.orderCount} 张订单，" f"冲突 {result.conflictCount} 项，总延期 {round(result.totalTardiness)} 小时。" f"可用检查点回滚到采用前。") if action in ("forecast.upsert", "forecast.delete", "forecast.convert"): from server.aps_domain.forecast import apply_forecast_action cp = _create_checkpoint(store, label="预测变更前基线", reason=f"auto:{action}", conversation_note=f"批准执行 {action}") applied = apply_forecast_action(store.data, store.next_id, action, params) fc = applied["forecast"] write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action=action, target={"type": "FORECAST_ORDER", "id": fc.get("id"), "forecastNo": fc.get("forecastNo")}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "beforeStatus": applied.get("beforeStatus"), "convertedOrderNo": applied.get("convertedOrderNo")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() if action == "forecast.convert": return (f"预测已转正 ✅ {fc.get('forecastNo')} → 销售订单 {applied.get('convertedOrderNo')}，" "可参与正式排产。") if action == "forecast.delete": return f"预测 {fc.get('forecastNo')} 已删除 ✅" verb = "已新建" if applied.get("created") else "已更新" return f"{verb}预测 {fc.get('forecastNo')} ✅（默认不进正式排产，可说「预测纳入试排」）。" if action == "data.import": # ---- 批准：批量导入 ---- from server.aps_domain.intake import apply_import cp = _create_checkpoint(store, label="批量导入前基线", reason="auto:data.import", conversation_note=f"批准导入 {params.get('kind')}") applied = apply_import(store.data, store.next_id, params) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="data.import", target={"type": "IMPORT", "id": params.get("kind")}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "count": applied["count"]}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() label = "订单" if applied["kind"] == "orders" else "物料" sample = "、".join(applied["created"][:5]) more = f" 等 {applied['count']} 条" if applied["count"] > 5 else f"共 {applied['count']} 条" return f"已导入{label} ✅ {sample}{more}。" if action == "flex.site.load": # ---- 批准：现场生产路线 ---- from server.importers import import_site_excel cp = _create_checkpoint(store, label="现场路线导入前基线", reason="auto:flex.site.load", conversation_note="批准加载完整生产路线") meta = import_site_excel( store.data, profile_id=params.get("profileId") or "kangni", route_path=params.get("routePath") or None, data_dir=params.get("dataDir") or None, include_sibling_orders=params.get("includeSiblings", False) is True, station_count=int(params.get("stationCount") or 0) or None, ) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="flex.site.load", target={"type": "FLEX_SITE", "id": meta.get("primaryOrder")}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "meta": meta}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return (f"现场生产路线已加载 ✅ 主单 {meta.get('primaryOrder')}，" f"订单 {meta.get('orderCount')} / 工序 {meta.get('operationCount')} / BOM {meta.get('bomCount')}。" f"演示 flex* 已清空。可说「柔性排产」。") if action == "import.commit": # ---- 批准：Excel/CSV 入库 ---- from server.aps_domain.importers import apply_import_commit cp = _create_checkpoint(store, label="文件导入前基线", reason="auto:import.commit", conversation_note=f"批准导入 {params.get('filename')}") batches = params.get("batches") or [] sql_payload = params.get("sqlPayload") mom_path = str(params.get("momPath") or "").strip() if (isinstance(sql_payload, dict) and sql_payload) or mom_path: # SQL/MOM 采用是整表替换：必须绑定出卡时的目标主数据快照， # 审批窗口内目标世界有任何业务改动都拒绝写入（与 folder.schedule 同口径）。 from server.aps_domain.folder_pack import folder_schedule_world_fingerprint expected_world = str(params.get("targetWorldFingerprint") or "").strip() if not expected_world or folder_schedule_world_fingerprint(store.data) != expected_world: raise PermissionError("采用目标主数据已变化，本次未写入；请重新分析后再采用") try: if isinstance(sql_payload, dict) and sql_payload: # 只读分析冻结的 SQL 载荷：采用即整包写入（不再重读来源文件）。 from server.importers.sql_pack import apply_sql_pack_to_world frozen_sql = copy.deepcopy(sql_payload) if _sql_payload_row_count(frozen_sql) <= 0: raise PermissionError( "SQL 数据包不含可采用的业务行，本次未写入；请重新分析后再采用") counts = apply_sql_pack_to_world(store.data, frozen_sql, replace=True) applied = {"summary": counts, "total": sum( int(value) for value in counts.values() if isinstance(value, (int, float)) and not isinstance(value, bool))} elif mom_path: # MOM 主数据表按文件采用：先按字节校验指纹，再用同一份字节解析 # （避免 hash→重开之间的替换竞态）；空表/无业务行一律拒绝写入。 expected = str(params.get("momSha256") or "").strip() try: raw = Path(mom_path).read_bytes() except OSError as exc: raise PermissionError( "来源文件不可读，本次未写入；请重新分析后再采用") from exc if not expected or hashlib.sha256(raw).hexdigest() != expected: raise PermissionError("来源文件已变化，本次未写入；请重新分析后再采用") from server.importers.mom_pack import ( apply_mom_pack_to_world, mom_pack_is_adoptable, parse_mom_workbook, ) with tempfile.TemporaryDirectory(prefix="aps-mom-adopt-") as frozen_dir: frozen_path = Path(str(frozen_dir)) / "frozen-source.xlsx" frozen_path.write_bytes(raw) pack = parse_mom_workbook(str(frozen_path)) if not mom_pack_is_adoptable(pack): raise PermissionError( "来源文件不足以作为 MOM 主数据采用（模型表或主干行不足），" "本次未写入；请重新分析后再采用") counts = apply_mom_pack_to_world(store.data, pack, replace=True) applied = {"summary": counts, "total": sum(counts.values())} elif params.get("intakeRecovery"): from server.aps_domain.intake_recovery import apply_reconciled_intake applied = apply_reconciled_intake(store.data, store.next_id, batches, params["intakeRecovery"]) else: applied = apply_import_commit(store.data, store.next_id, batches) except (ValueError, PermissionError) as exc: if not (params.get("sourceProfile") or any(batch.get("sourceProfile") for batch in batches)): raise from server.aps_domain.planning_intake import PlanningIntakeRejected write_audit(store.data, store.next_id, actor=actor, category="GATE", action="import.commit.denied", target={"type": "IMPORT", "id": params.get("filename")}, power="P2", rationale={"confirmId": confirm_id, "reason": str(exc)}, result="DENIED", evidence_refs=evidence_refs) store.save() raise PlanningIntakeRejected(str(exc)) from exc # 采用即归属：只补首次归属，后续审批人不会改写已登记的数据owner。 try: from server.knowledge.ingest import bind_world_to_identity if not (store.data.get("meta") or {}).get("ownerUserId"): bind_world_to_identity( store.data, sources=[str(params.get("filename") or "")]) except Exception: # noqa: BLE001 - 归属登记失败不阻断已批准的采用 pass write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="import.commit", target={"type": "IMPORT", "id": params.get("filename")}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "summary": applied["summary"]}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() ingest_meta = params.get("analysisIngest") if isinstance(ingest_meta, dict): # 分析报告入租户知识库：只在采用获批后发生，并带上已落库的完整数据集。 try: from server.knowledge.ingest import ingest_project_analyze_report project = str(ingest_meta.get("projectName") or params.get("filename") or "当前项目") counts = applied.get("summary") or {} detail = "、".join(f"{k} {v}" for k, v in counts.items()) or "无新增行" ingest_project_analyze_report({ "projectName": project, "summary": counts, "sources": ingest_meta.get("sources") or [params.get("filename")], "sourcePaths": ingest_meta.get("sourcePaths") or [], "plan": [], "markdown": (f"# {project} 资料采用\n\n" f"- 来源：{'、'.join(str(s) for s in (ingest_meta.get('sources') or [])) or params.get('filename')}\n" f"- 采用结果：{detail}\n" f"- 采用人：{actor}\n"), }, world=store.data) except Exception: # noqa: BLE001 - 知识入库失败不回滚已批准的采用 pass from server.importers.workbook_profiles import has_adoption_flow # 采用流：注册格式的采用卡，或由只读分析(analysisIngest)升级出的采用卡； # 两者都走人工采用的措辞，避免把“采用资料”说成普通文件导入。 if has_adoption_flow(params.get("sourceProfile")) or isinstance(ingest_meta, dict): recovered = applied.get("recoveredOrderNos") or [] note = f"已将 {'、'.join(recovered)} 恢复为待评估插单，原记录已归档。" if recovered else "" adopted = "、".join(f"{k}×{v}" for k, v in (applied.get("summary") or {}).items()) or "无新增行" return (note + f"排产资料已采用（{adopted}），尚未生成方案。你可以在主数据中核对和修改，" "再说「根据这些数据排产」。原表的来源说明与待确认信息已保留。") detail = "、".join(f"{k}×{v}" for k, v in applied["summary"].items()) or "无行" return f"文件导入完成 ✅ 共 {applied['total']} 条（{detail}）。" if action == "folder.schedule": from server.aps_domain.importers import apply_import_commit from server.importers.sql_pack import apply_sql_pack_to_world candidate = _FolderCandidateStore(store) batches = copy.deepcopy(params.get("batches") or []) sql_payload = copy.deepcopy(params.get("sqlPayload")) kangni_payload = copy.deepcopy(params.get("kangniPayload")) kangni_detected = params.get("kangniDetected") is True if kangni_detected: batches = [] sql_payload = None sql_already = isinstance(sql_payload, dict) and bool(sql_payload) applied = {"total": 0, "summary": {}} applied_meta: dict[str, Any] = {} resource_quality: dict[str, Any] = {} try: if kangni_detected: from server.aps_domain.kangni_intake import apply_site_payload_to_world applied_meta = apply_site_payload_to_world( candidate.data, kangni_payload, clear_all=True, ) for sales_order in candidate.data.get("salesOrders") or []: sales_order["source"] = "SITE" resource_quality = copy.deepcopy( ((kangni_payload or {}).get("meta") or {}).get("resourceQuality") or {} ) applied = { "total": int(applied_meta.get("orderCount") or 0), "summary": { "salesOrders": len(candidate.data.get("salesOrders") or []), "flexOrders": len(candidate.data.get("flexOrders") or []), "flexRoutings": len(candidate.data.get("flexRoutings") or []), "flexEquipment": len(candidate.data.get("flexEquipment") or []), "flexMolds": len(candidate.data.get("flexMolds") or []), }, } elif sql_already: apply_sql_pack_to_world(candidate.data, sql_payload, replace=True) if batches and not kangni_detected: applied = apply_import_commit(candidate.data, candidate.next_id, batches) from server.aps_domain.readiness import check_readiness, readiness_text ready = check_readiness(candidate.data) has_core = ( bool(candidate.data.get("flexOrders")) and bool(candidate.data.get("flexRoutings")) and any( e.get("status") == "RUNNING" for e in (candidate.data.get("flexEquipment") or []) ) ) if ready["summary"].get("ready", 0) == 0 and not has_core: raise ValueError("导入后仍不具备开排条件：" + readiness_text(ready)) from server.aps_domain.flex import run_flex_schedule result = run_flex_schedule( candidate, sort_mode=params.get("sortMode") or "BOTTLENECK", actor=actor, trial=True, ) production_ready = bool(kangni_detected and resource_quality.get("productionReady")) result["trialOnly"] = True result["productionReady"] = production_ready if kangni_detected: result["resourceQuality"] = copy.deepcopy(resource_quality) version_id = result.get("versionId") for version in candidate.data.get("flexScheduleVersions") or []: if version.get("id") == version_id: version["trialOnly"] = True version["productionReady"] = production_ready if kangni_detected: version["resourceQuality"] = copy.deepcopy(resource_quality) break write_audit( candidate.data, candidate.next_id, actor=actor, category="WORLD_WRITE", action="folder.schedule.import", target={"type": "FOLDER_PACK", "id": params.get("projectId") or "project"}, power="P2", rationale={ "confirmId": confirm_id, "approver": actor, "summary": applied.get("summary"), "sqlApplied": sql_already, "kangniDetected": kangni_detected, "trialOnly": True, "productionReady": ( bool(resource_quality.get("productionReady")) if kangni_detected else None ), "sourceManifestDigest": params.get("sourceManifestDigest"), "folderPayloadDigest": params.get("folderPayloadDigest"), "frozenPayload": True, }, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs, ) except Exception as exc: # noqa: BLE001 - candidate world is discarded below try: from server.agent_core.plan_orchestration import decide_plan_node decide_plan_node( confirm_id=confirm_id, approve=False, note=f"folder execution denied: {exc}", actor=actor, ) except Exception: pass write_audit( store.data, store.next_id, actor=actor, category="GATE", action="folder.schedule.execute.denied", target={"type": "FOLDER_PACK", "id": params.get("projectId") or "project"}, power="P2", rationale={"confirmId": confirm_id, "reason": str(exc)}, result="DENIED", before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs, ) store.save() return f"工程目录导入与试排失败，已回滚且未写入业务数据：{exc}" previous_world = store.data previous_counter_state = { attr: copy.deepcopy(getattr(store, attr)) for attr in ("_counters", "_ids") if hasattr(store, attr) } store.data = candidate.data try: reset_counters = getattr(store, "_reset_counters", None) if callable(reset_counters): reset_counters() else: committed_counters = copy.deepcopy(getattr(candidate, "_counters", {})) for attr, previous in previous_counter_state.items(): synchronized = copy.deepcopy(previous) for kind, value in committed_counters.items(): synchronized[kind] = max(int(synchronized.get(kind) or 0), int(value)) setattr(store, attr, synchronized) store.save() except Exception as exc: # noqa: BLE001 - restore old in-memory world after atomic write failure store.data = previous_world for attr, previous in previous_counter_state.items(): setattr(store, attr, previous) try: from server.agent_core.plan_orchestration import decide_plan_node decide_plan_node( confirm_id=confirm_id, approve=False, note=f"folder save denied: {exc}", actor=actor, ) except Exception: pass write_audit( store.data, store.next_id, actor=actor, category="GATE", action="folder.schedule.save.denied", target={"type": "FOLDER_PACK", "id": params.get("projectId") or "project"}, power="P2", rationale={"confirmId": confirm_id, "reason": str(exc)}, result="DENIED", before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs, ) try: store.save() except Exception: # noqa: BLE001 - original disk image remains authoritative pass return f"工程目录导入与试排落盘失败，已恢复原业务状态：{exc}" try: from server.agent_core.plan_orchestration import decide_plan_node decide_plan_node(confirm_id=confirm_id, approve=True, note=note, actor=actor) except Exception: pass # The validated candidate already contains the schedule we just committed. # Running again here creates a second version outside the atomic transaction. head = "工程目录已导入。" ver = (result or {}).get("versionNo") or "" return (head + f"\n已按目录数据生成试排方案：**{ver}**" f"（生产分组 {(result or {}).get('vlCount', 0)} 个 · " f"待处理问题 {(result or {}).get('conflictCount', 0)} 项）。" "\n这一步尚未下发到车间。请查看右侧排产结果，核对订单、设备和交期；" "也可以查看冲突或导出排产表。") if action == "flex.reschedule": from server.aps_domain.flex import reschedule_by_level cp = _create_checkpoint(store, label=f"{params.get('level', 'L2')} 重排前基线", reason="auto:flex.reschedule", conversation_note=f"批准 {params.get('level')} 重排") r = reschedule_by_level(store, level=params.get("level") or "L2", sort_mode=params.get("sortMode"), actor=actor) sch = r["schedule"] # EX-03：若从冲突修复卡进入，标记原冲突已解决 cid = params.get("resolveConflictId") if cid is not None: for cf in store.data.get("flexConflicts", []): if cf.get("id") == cid: cf["isResolved"] = True cf["resolutionAction"] = f"flex.reschedule:{r['level']}" cf["resolvedBy"] = actor break write_audit(store.data, store.next_id, actor=actor, category="GATE", action="flex.reschedule.approve", target={"type": "FLEX_VERSION", "id": sch.get("versionId")}, power="P2", rationale={"confirmId": confirm_id, "level": r["level"], "frozenWo": r.get("frozenWo"), "mutableOrders": r.get("mutableOrders"), "resolveConflictId": cid}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return (f"{r['level']} 重排已执行 ✅ 版本 {sch.get('versionNo')}：" f"冻结 {r.get('frozenWo', 0)} / 重排 {r.get('mutableOrders', 0)} 单，" f"冲突 {sch.get('conflictCount', 0)}。") if action == "flex.adjust.commit": from server.aps_domain.adjust import commit_flex_adjust cp = _create_checkpoint(store, label="甘特调程前基线", reason="auto:flex.adjust.commit", conversation_note=f"批准拖拽工单 #{params.get('woId')}") r = commit_flex_adjust(store, int(params["woId"]), str(params["newStart"]), actor=actor) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="flex.adjust.approve", target={"type": "FLEX_VERSION", "id": r.get("versionId")}, power="P2", rationale={"confirmId": confirm_id, "woId": params.get("woId")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return r.get("message") or "调程已执行 ✅" if action == "schedule.adjust.commit": from server.aps_domain.adjust import commit_fixed_adjust cp = _create_checkpoint(store, label="固定甘特调程前基线", reason="auto:schedule.adjust.commit", conversation_note=f"批准拖拽工单 #{params.get('woId')}") r = commit_fixed_adjust(store, int(params["woId"]), str(params["newStart"]), actor=actor) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="schedule.adjust.approve", target={"type": "SCHEDULE_VERSION", "id": r.get("versionId")}, power="P2", rationale={"confirmId": confirm_id, "woId": params.get("woId")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return r.get("message") or "调程已执行 ✅" if action == "sap.sync.inbound": from server.aps_domain.sap_sync import apply_inbound cp = _create_checkpoint(store, label="SAP 入站前基线", reason="auto:sap.sync.inbound", conversation_note="批准 SAP 入站同步") r = apply_inbound(store, actor=actor) return r.get("message") or "SAP 入站完成 ✅" if action == "sap.sync.outbound": from server.aps_domain.sap_sync import apply_outbound # Pass the frozen decision fields verbatim; never create a final-stage snapshot. try: r = apply_outbound( store, actor=actor, confirm_id=confirm_id, approval_params=params, execution_grant=pending.get("executionGrant"), requester=pending.get("requester"), approvals=pending.get("approvals"), before_snapshot=pending.get("beforeSnapshot"), checkpoint_store=get_checkpoints(), ) except PermissionError as exc: return f"SAP 出站证据校验未通过，未执行外部回写：{exc}" return r.get("message") or "SAP 出站完成 ✅" if action == "mes.dispatch": from server.aps_domain.mes import apply_dispatch checkpoints = get_checkpoints() checkpoint = checkpoints.create( store.data, label="MES 下发前基线", reason="auto:mes.dispatch", conversation_note=f"批准 MES 下发 {params.get('track')}", ) r = apply_dispatch( store, track=params.get("track") or "flex", actor=actor, confirm_id=confirm_id, execution_grant=str(pending.get("executionGrant") or ""), version_id=int(params.get("versionId")), before_snapshot=str(checkpoint["pairId"]), evidence_refs=list(params.get("evidenceRefs") or []), checkpoint_store=checkpoints, ) return r.get("message") or "MES 下发完成 ✅" if action == "mrp.release": # ---- 批准：MRP 建议单下达 ---- from server.aps_domain.mrp import apply_mrp_release cp = _create_checkpoint(store, label="MRP 下达前基线", reason="auto:mrp.release", conversation_note="批准下达 MRP 建议单") # P2 写前自动建档 applied = apply_mrp_release(store.data, params.get("orderNo"), params.get("kind", "all")) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="mrp.release", target={"type": "MRP", "id": params.get("orderNo") or "ALL"}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "purchase": applied["purchaseCount"], "outsource": applied["outsourceCount"]}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return (f"已下达 ✅ 采购单 {applied['purchaseCount']} 条 / 委外单 {applied['outsourceCount']} 条转为正式（RELEASED），" "可在订单面板「分解建议」查看状态。") if action.startswith("master."): # ---- 批准：主数据写入（白名单见 masterdata.MASTER_ACTIONS） ---- from server.aps_domain.masterdata import apply_master_action cp = _create_checkpoint(store, label="主数据变更前基线", reason=f"auto:{action}", conversation_note=f"批准执行 {action}") # P2 写前自动建档 applied = apply_master_action(store.data, store.next_id, action, params) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action=action, target={"type": f"MASTER_{applied['kind']}", "id": applied["id"], "name": applied["name"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "before": applied.get("beforeStatus", applied.get("beforeStock")), "after": applied.get("afterStatus", applied.get("afterStock"))}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() kind_cn = { "LINE": "产线", "MATERIAL": "物料", "MAINTENANCE": "维保计划", "BOM_ITEM": "BOM 明细", "BOM": "BOM", "ROUTING_STEP": "工艺步骤", "ROUTING": "工艺路线", "OPERATION": "工序", "LINE_PRODUCT": "产线绑定", "MASTER_CLEAR": "主数据清理", }.get(applied["kind"], applied["kind"]) if applied["kind"] == "MASTER_CLEAR": detail = ", ".join(applied.get("cleared") or applied.get("restored") or []) or "无数据" return f"主数据已按【{applied['name']}】清空 ✅ {detail}（变更前已自动建档）。" return f"{kind_cn}【{applied['name']}】主数据已更新 ✅ 只影响后续新排产版本（变更前已自动建档）。" if action == "params.update": # ---- 批准：排产参数（OR-02） ---- from server.aps_domain.params import apply_params_update cp = _create_checkpoint(store, label="排产参数变更前基线", reason="auto:params.update", conversation_note="批准更新排产参数") applied = apply_params_update(store.data, params) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="params.update", target={"type": "SCHEDULE_PARAMS", "id": "scheduleParams"}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "before": applied["before"].get("customerLevelWeights"), "after": applied["after"].get("customerLevelWeights"), "reset": applied.get("reset")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() lw = applied["after"]["customerLevelWeights"] return (f"排产参数已更新 ✅ 客户等级权重 VIP={lw['VIP']} / A={lw['A']} / B={lw['B']} / C={lw['C']}。" "只影响后续新排产版本（变更前已自动建档）。") if action == "param.experiment.promote": # ---- 批准：参数实验升级（矩阵 87 P2 门禁） ---- from server.agent_core.param_opt import DEFAULT_TOLERANCE, ParameterOptimizer experiment_id = str(params.get("experimentId") or "") existing = next((e for e in store.data.get("paramExperiments") or [] if e.get("id") == experiment_id), None) if existing is None: return f"未找到参数实验：{experiment_id}，未做任何变更。" if existing.get("status") != "GRAY": return (f"参数实验 {experiment_id} 状态为 {existing.get('status')}，" "只能从 GRAY 升级，未做任何变更。") cp = _create_checkpoint(store, label="参数实验升级前基线", reason="auto:param.experiment.promote", conversation_note="批准参数实验 GRAY→ACTIVE") opt = ParameterOptimizer( tolerance=float(existing.get("tolerance") or DEFAULT_TOLERANCE), strategy=existing.get("strategy") or "COMPREHENSIVE") rec = opt.finalize(store.data, experiment_id, promote_full=True) v_replay = rec.get("validationReplay") or {} write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="param.promote", target={"type": "PARAM_EXPERIMENT", "id": experiment_id}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "status": rec.get("status"), "rollbackReason": rec.get("rollbackReason"), "validationReplay": { "degraded": v_replay.get("degraded"), "baselineKpi": v_replay.get("baselineKpi"), "candidateKpi": v_replay.get("candidateKpi"), "deltas": v_replay.get("deltas"), }}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() if rec.get("status") == "ROLLED_BACK": return (f"参数实验 {experiment_id} 已批准升级，但验证集回放退化，" f"已自动回滚并保持基线参数（{rec.get('rollbackReason')}）。") status_cn = "全量生效（FULL）" if rec.get("status") == "FULL" else "ACTIVE" return (f"参数实验 {experiment_id} 已升级为 {status_cn} ✅ " "候选参数经验证集回放通过后生效（升级前已自动建档）。") if action == "constraint.profile.save": # ---- 批准：约束剖面（SC-04） ---- from server.aps_domain.constraints import apply_profile_save cp = _create_checkpoint(store, label="约束配置变更前基线", reason="auto:constraint.profile.save", conversation_note="批准更新约束配置") applied = apply_profile_save(store.data, params) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="constraint.profile.save", target={"type": "CONSTRAINT_PROFILE", "id": applied["id"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "reset": applied.get("reset")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() enabled = sum(1 for c in applied["after"]["constraints"] if c["enabled"]) hard_n = sum(1 for c in applied["after"]["constraints"] if c["kind"] == "hard" and c["enabled"]) return (f"约束配置已更新 ✅ 启用 {enabled} 条 / 其中硬约束 {hard_n} 条。" "硬约束违反将阻止发布（变更前已自动建档）。") if action == "constraint.rule.save": # ---- 批准：计划员扩充规则（SC-04） ---- from server.aps_domain.constraint_rules import apply_rule_save cp = _create_checkpoint(store, label="约束规则库变更前基线", reason="auto:constraint.rule.save", conversation_note="批准扩充约束规则") applied = apply_rule_save(store.data, params, getattr(store, "tenant_uuid", "platform")) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="constraint.rule.save", target={"type": "CONSTRAINT_RULE", "id": applied["ruleId"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "op": applied["op"], "scope": applied.get("scope")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return str(applied.get("message") or f"约束规则 {applied['ruleId']} 已更新 ✅") if action == "sop.apply": # ---- 批准：SOP 规则包（IND-02） ---- from server.aps_domain.sop_rules import apply_sop_pack, compile_sop_by_asset pack = params.get("pack") if not isinstance(pack, dict): asset_id = str(params.get("assetId") or params.get("asset_id") or "").strip() if not asset_id: return "应用 SOP 被拒绝：确认卡缺少 assetId，且没有预编译规则包。" compiled = compile_sop_by_asset(get_knowledge().assets, asset_id) if not compiled.get("ok"): return compiled.get("error") or "SOP 编译失败" pack = compiled["pack"] cp = _create_checkpoint(store, label="SOP规则包应用前基线", reason="auto:sop.apply", conversation_note="批准应用 SOP→约束") applied = apply_sop_pack(store.data, pack) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="sop.apply", target={"type": "RULE_PACK", "id": applied["packId"]}, power="P2", rationale={"confirmId": confirm_id, "approver": actor, "applied": applied.get("applied"), "source": pack.get("source")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() note = "；".join(applied.get("notes") or []) or "无额外流程提示" return ( f"已应用 SOP 规则包 ✅ {pack.get('title')}（{applied['packId']}）。\n" f"生效项：{', '.join(applied.get('applied') or [])}\n{note}" ) if action == "knowledge.import": from server.knowledge.embedding import index_units from server.knowledge.ingest import apply_knowledge_import, preview_ingest path = params.get("path") if not path: return "缺少文件路径，无法导入知识文档" try: preview = preview_ingest( params.get("filename") or path, path=path, kind=params.get("kind") or "sop", title=params.get("title"), ) applied = apply_knowledge_import(preview) # 尝试建向量索引（无嵌入后端则跳过） # 复用模块级 get_knowledge（顶部已导入），避免局部 import 遮蔽导致 F823/NameError index_units(get_knowledge().iter_search_units()) except (FileNotFoundError, ValueError) as exc: return str(exc) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="knowledge.import", target={"type": "KNOWLEDGE", "id": applied["assetId"]}, power="P2", rationale={"confirmId": confirm_id, "chunkCount": applied.get("chunkCount"), "title": applied.get("title")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return (f"知识文档已入库 ✅ 《{applied['title']}》{applied['version']}，" f"切块 {applied['chunkCount']} 段（资产 {applied['assetId']}）。") if action == "skill.register": from server.agent_core.skills import get_skills payload = { "skill_id": params.get("skill_id") or params.get("skillId"), "name": params.get("name") or params.get("skill_id") or params.get("skillId"), "endpoint": params.get("endpoint"), "auth": params.get("auth") or "", "track": params.get("track") or "flex", "enabled": params.get("enabled", True), "description": params.get("description") or "", } saved = get_skills().upsert(payload) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="skill.register", target={"type": "SKILL", "id": saved["skill_id"]}, power="P2", rationale={"confirmId": confirm_id, "endpoint": saved.get("endpoint")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return f"已登记算法 Skill ✅ {saved['skill_id']}（{saved['name']}）→ {saved['endpoint']}" if action == "skill.enable": from server.agent_core.skills import get_skills sid = params.get("skill_id") or params.get("skillId") enabled = params.get("enabled") if not isinstance(enabled, bool): return "确认卡缺少布尔 enabled，未执行算法 Skill 启停。" saved = get_skills().set_enabled(sid, enabled) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="skill.enable", target={"type": "SKILL", "id": sid}, power="P2", rationale={"confirmId": confirm_id, "enabled": enabled}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return f"算法 Skill {sid} 已{'启用' if enabled else '停用'} ✅" if action == "flex.time.update": # ---- 批准：工时维护 ---- op_code = str(params.get("operationCode") or "") op_codes = set(params.get("operationCodes") or ([op_code] if op_code else [])) pc = str(params.get("productCode") or "") std_min = float(params.get("stdMin") or 0) source = params.get("source") or "实测" # 写前快照（统一证据链：P2 写世界必须有回滚锚点） cp = _create_checkpoint(store, label="工时维护前基线", reason="auto:flex.time.update", conversation_note="批准更新工时") hits = 0 for step in store.data.get("flexRoutings") or []: if op_codes and step.get("operationCode") not in op_codes: continue if pc and step.get("productCode") != pc: continue step["stdTimePerUnit"] = std_min step["stdTimeSource"] = source hits += 1 # 同步工序库标准工时（主数据页显示口径） for op in store.data.get("flexOperations") or []: if op.get("code") in op_codes and not pc: op["stdTimeMin"] = std_min write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="flex.time.update", target={"type": "ROUTING_TIME", "id": f"{pc or '*'}/{op_code or '*'}"}, power="P2", rationale={"confirmId": confirm_id, "stdMin": std_min, "hits": hits, "source": source}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() ops_label = "、".join(sorted(op_codes)) or "全部工序" if not hits: return f"未找到匹配的路线步骤（产品={pc or '任意'}，工序={ops_label}），未做修改。" return (f"工时已更新 ✅ {pc or '全部产品'} × {ops_label} → " f"{std_min} 分钟/件（来源={source}，{hits} 条路线步骤）。") if action == "routing.template.apply": # ---- 批准：模板生成工艺路线 ---- from server.knowledge.routing_templates import apply_template_to_product cp = _create_checkpoint(store, label="模板生成路线前基线", reason="auto:routing.template.apply", conversation_note=f"模板 {params.get('templateCode')} → {params.get('productCode')}") try: applied = apply_template_to_product( store.data, str(params.get("templateCode") or ""), str(params.get("productCode") or ""), str(params.get("productName") or "")) except ValueError as exc: return str(exc) write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="routing.template.apply", target={"type": "ROUTING", "id": applied["productCode"]}, power="P2", rationale={"confirmId": confirm_id, "template": applied["template"], "steps": applied["steps"], "assetId": applied.get("assetId")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return (f"工艺路线已生成 ✅ {applied['productCode']} ← 模板「{applied['templateName']}」" f"（{applied['steps']} 步，工时标「模板」，出处 {applied.get('assetId') or '内置'}）。" f"可说「数据齐备度」复查后「柔性排产」。") if action == "drawing.master.apply": drawing_id = str(params.get("drawingId") or "") selected = list(params.get("selectedCandidates") or []) source_hash = str(params.get("sourceSha256") or "") created = 0 linked = 0 materials = store.data.setdefault("materials", []) links = store.data.setdefault("drawingLinks", []) from server.aps_domain.drawing_understanding import build_drawing_link_evidence_records, utc_now # noqa: I001 for candidate in selected: if candidate.get("candidateType") != "MATERIAL": continue code = str(candidate.get("code") or "").strip() name = str(candidate.get("name") or "").strip() if not code or not name: continue row = next((m for m in materials if str(m.get("code") or "") == code), None) if row is None: row = {"id": store.next_id("material"), "code": code, "name": name, "spec": "", "type": "SEMI_FINISHED", "unit": "piece", "stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0, "productFamily": "DRAWING_PART", "status": "ACTIVE", "source": "DXF_P2_CONFIRMED"} materials.append(row) created += 1 link_key = (drawing_id, row.get("id"), source_hash) if not any((x.get("drawingId"), x.get("materialId"), x.get("sourceSha256")) == link_key for x in links): links.append({"id": store.next_id("drawingLink"), "drawingId": drawing_id, "materialId": row.get("id"), "linkType": "MATERIAL", "targetType": "material", "targetId": row.get("id"), "targetRef": code, "sourceSha256": source_hash, "revision": candidate.get("revision"), "status": "CONFIRMED", "masterCommitted": True, "confidence": round(min(float(candidate.get("confidence") or 0.0), 1.0), 3), "confirmedBy": actor, "confirmedAt": utc_now(), "evidence": {"contractVersion": "drawing-link-evidence.v1", "confirmId": confirm_id, "candidateId": candidate.get("candidateId"), "candidateType": candidate.get("candidateType"), "code": code, "sourceSha256": source_hash, "drawingId": drawing_id}}) linked += 1 evidence_records = build_drawing_link_evidence_records( store.data, drawing_id=drawing_id, source_sha256=source_hash, selected_candidates=selected, actor=actor, confirm_id=confirm_id, next_id=store.next_id, ) for link in evidence_records: link_key = (drawing_id, link.get("linkType"), link.get("targetType"), link.get("targetRef"), source_hash) if not any((x.get("drawingId"), x.get("linkType"), x.get("targetType"), x.get("targetRef"), x.get("sourceSha256")) == link_key for x in links): links.append(link) linked += 1 for change in store.data.get("drawingCandidates") or []: if change.get("drawingAssetId") == drawing_id: change["status"] = "PARTIALLY_COMMITTED" if created or linked else "REVIEW_CONFIRMED" change["confirmedCandidateIds"] = [x.get("candidateId") for x in selected] for asset in store.data.get("drawingAssets") or []: if asset.get("id") == drawing_id: asset["status"] = "COMMITTED" if created or linked else "REVIEW_REQUIRED" write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="drawing.master.apply", target={"type": "DRAWING", "id": drawing_id}, power="P2", rationale={"confirmId": confirm_id, "sourceSha256": source_hash, "materialCreated": created, "linksCreated": linked, "deferredIncompleteCandidates": sum(1 for x in selected if x.get("candidateType") != "MATERIAL")}, before_snapshot=str(cp["pairId"]) if cp else None, evidence_refs=evidence_refs) store.save() return (f"Drawing review applied: materials={created}, links={linked}; " "incomplete BOM/routing candidates remain pending and were not fabricated.") return "未知动作，已忽略。" # 白名单外兜底（理论不可达） # ---------------- 柔性排产（M5：能力池 + 虚拟产线 / 瓶颈产能 / 交期承诺） ---------------- _FLEX_MODE_CN = {"ASC": "正排(交期优先)", "DESC": "倒排(交期倒推)", "BOTTLENECK": "瓶颈锚"} def _flex_product_name(store: WorldStore, code: str) -> str: """物料/产品编码 → 中文名（回复文案友好化）。""" m = next((x for x in store.data.get("flexMaterials", []) if x["code"] == code), None) return f"{m['name']}({code})" if m else code def _resolve_order_reference(store: WorldStore, session_id: str, intent: IntentResult) -> IntentResult: """把 orderRef=last / 口令中的订单号 解析为 flex orderIds 或 orderNo。""" from server.agent_core.session_focus import get_last_order_no, set_focus params = dict(intent.params or {}) wanted_ref = params.get("orderRef") in ("last", "this", "focus") order_no = str(params.get("orderNo") or "").strip() if not order_no and wanted_ref: order_no = get_last_order_no(session_id) or "" if not order_no: # 世界里只有一单时自动锁定（现场常见：刚导入的主单） flex = store.data.get("flexOrders") or [] sales = store.data.get("salesOrders") or [] if len(flex) == 1: order_no = str(flex[0].get("orderNo") or "") elif len(sales) == 1: order_no = str(sales[0].get("orderNo") or "") if order_no: params.pop("_missingFocus", None) params["orderNo"] = order_no set_focus(session_id, orderNo=order_no) fo = next((o for o in (store.data.get("flexOrders") or []) if str(o.get("orderNo")) == order_no), None) if fo and intent.intent == "flex.schedule": params["orderIds"] = [fo["id"]] elif wanted_ref: # 「这个订单」但会话未锁定、且池内多单 → 不瞎排全池 params["_missingFocus"] = True return IntentResult(intent=intent.intent, params=params, confidence=intent.confidence, source=intent.source) def _external_inputs_hash(world: dict, summary: dict) -> str | None: """EXTERNAL 引擎输入包指纹（可复算）：柔性订单数 + 产品 + skillId 稳定投影。""" import hashlib, json as _json try: payload = { "flexOrders": len(world.get("flexOrders") or []), "products": sorted({p.get("code") or p.get("name") or "" for p in (world.get("products") or [])}), "skillId": summary.get("skillId") or "", } return hashlib.sha256( _json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() except Exception: return None def _build_external_trace(store: WorldStore, summary: dict) -> dict: """构造 flex.schedule.external 可追溯链（尽力而为；失败返回空链摘要不影响排产）。 链哈希只含确定性信息：版本/run/EXTERNAL/输入包指纹/结果计数； skill 上游墙钟等不确定量不进入链哈希。 """ try: from server.agent_core.evidence import EvidenceItem, trace_chain, evidence_ref version_ref = evidence_ref("schedule-version", summary["versionId"]) run_ref = evidence_ref("run", summary.get("runId") or summary["versionId"]) items = [ EvidenceItem(kind="schedule-version", ref=str(summary["versionId"]), version=str(summary["versionId"])), EvidenceItem(kind="run", ref=run_ref, runId=str(summary.get("runId") or summary["versionId"]), engine="EXTERNAL", meta={"skillId": summary.get("skillId") or "", "conflictCount": summary.get("conflictCount"), "avgUtilization": summary.get("avgUtilization"), "onTimeCount": summary.get("onTimeCount")}), EvidenceItem(kind="algorithm", ref=summary.get("skillId") or "EXTERNAL", version=summary.get("skillId") or "EXTERNAL", inputsHash=_external_inputs_hash(store.data, summary)), ] chain = trace_chain(items) return { "chainHash": chain["chainHash"], "count": chain["count"], "summary": [{"kind": it["kind"], "ref": it["ref"]} for it in chain["items"]], "items": chain["items"], "versionRef": version_ref, } except Exception: return {"chainHash": None, "count": 0, "summary": [], "versionRef": None} def _external_audit_evidence(summary: dict, trace: dict) -> list[str]: """合并 EXTERNAL 结果引用（skill/run）与 trace 的 schedule-version 引用（去重保序）。""" refs = [] skill_id = summary.get("skillId") run_id = summary.get("runId") if skill_id: refs.append(f"skill:{skill_id}") if run_id: refs.append(f"run:{run_id}") version_ref = trace.get("versionRef") if version_ref and version_ref not in refs: refs.append(version_ref) return refs def _run_flex(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply: """触发柔性排产并回执（短文案 + 结构化 flex-schedule 块：KPI + 虚拟产线表 + 瓶颈）。""" from server.aps_domain.flex import run_flex_schedule from server.importers.workbook_profiles import has_adoption_flow from server.timeutil import parse_dt planning_context = store.data.get("planningContext") or {} # Adopted workbooks are explicitly trial-only. Real blockers must stay # visible, but they must not prevent the draft trial that produces that # blocker list. Formal publication remains a separate P2 gate. if has_adoption_flow(planning_context.get("sourceProfile")) and not planning_context.get("trialOnly"): from server.aps_domain.readiness import check_readiness readiness = check_readiness(store.data) if int((readiness.get("summary") or {}).get("ready") or 0) <= 0: return AgentReply( text="当前资料未通过排产检查，未生成方案。请先补齐订单、工艺、工时、设备或班次，再开始排产。" ) mode = (intent.params.get("sortMode") or planning_context.get("defaultSortMode") or (store.data.get("flexParams") or {}).get("sortMode")) if not mode: return AgentReply(text="请先选择本次排产目标，再生成方案。") window = intent.params.get("window") order_ids = intent.params.get("orderIds") if isinstance(order_ids, list): order_ids = [int(x) for x in order_ids if str(x).isdigit() or isinstance(x, int)] else: order_ids = None engine = str(intent.params.get("engine") or "").upper() use_ext = bool(engine.startswith("EXTERNAL") or intent.params.get("skillId") or intent.params.get("useExternal")) if use_ext: from server.engines.external_engine import run_external_flex try: result = run_external_flex( store, skill_id=intent.params.get("skillId"), order_ids=order_ids, actor=actor, ) except ValueError as exc: return AgentReply(text=str(exc)) ext_trace = _build_external_trace(store, result) write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="flex.schedule.external", target={"type": "FLEX", "id": result.get("versionNo")}, power="P1", rationale={"skillId": result.get("skillId"), "runId": result.get("runId"), "traceChainHash": ext_trace["chainHash"], "traceCount": ext_trace["count"], "traceSummary": ext_trace["summary"], "traceItems": ext_trace.get("items"), "evidence": _external_audit_evidence(result, ext_trace)}) store.save() else: result = run_flex_schedule(store, sort_mode=mode, order_ids=order_ids or None, actor=actor, window=window) # P1 mode_key = str(result.get("sortMode") or result.get("engineType") or mode or "CLOSED_LOOP") mode_cn = _FLEX_MODE_CN.get(mode_key, mode_key) solve_status = str(result.get("solveStatus") or "FEASIBLE").upper() if result.get("sortMode") == "EXTERNAL" or use_ext: mode_cn = f"外部算法({result.get('skillId') or 'skill'})" win_cn = {"short": "短窗2h", "mid": "中窗2d", "long": "长窗7d", "full": "全量"}.get( result.get("window") or "full", result.get("window") or "full") vls = [v for v in store.data["flexVirtualLines"] if v["versionId"] == result["versionId"]] orders_by_no = {o["orderNo"]: o for o in store.data["flexOrders"]} def _on_time(v) -> bool: # 完工是否不晚于交期 o = orders_by_no.get(v["orderNo"]) return bool(o and v["plannedEnd"] and parse_dt(v["plannedEnd"]) <= parse_dt(o["dueDate"] + " 18:00")) wos_by_vl: dict[int, list] = {} # 工单按虚拟产线归集（下钻依据） for w in store.data["flexWorkOrders"]: if w["versionId"] == result["versionId"]: wos_by_vl.setdefault(w["vlId"], []).append(w) def _steps(vl_id: int) -> list[dict]: # 单条虚拟产线的工序级明细 rows = sorted(wos_by_vl.get(vl_id, []), key=lambda w: w["seq"]) return [{ "seq": w["seq"], "operationName": w["operationName"], "equipmentName": w["equipmentName"], "equipmentCode": w["equipmentCode"], "zone": w.get("zone"), "moldCode": w.get("moldCode"), "changeoverMin": w.get("changeoverMin", 0), "moveMin": w.get("moveMin", 0), "runMin": w.get("runMin", 0), "start": w["plannedStartTime"], "end": w["plannedEndTime"], "isBottleneck": w.get("isBottleneck", False), } for w in rows] lines = [{ "vlNo": v["vlNo"], "orderNo": v["orderNo"], "product": _flex_product_name(store, v["productCode"]), "quantity": v["quantity"], "start": v["plannedStart"], "end": v["plannedEnd"], "due": (orders_by_no.get(v["orderNo"]) or {}).get("dueDate"), "priority": (orders_by_no.get(v["orderNo"]) or {}).get("priority"), "stepCount": len(wos_by_vl.get(v["id"], [])), "onTime": _on_time(v), "status": "scheduled", "steps": _steps(v["id"]), # 工单级依据（点开可见） } for v in vls] # 未排出的订单也上表，避免「有 KPI、表却空」 scheduled_nos = {ln["orderNo"] for ln in lines} for c in store.data.get("flexConflicts") or []: if c.get("versionId") != result["versionId"]: continue ono = c.get("orderNo") if not ono or ono in scheduled_nos: continue o = orders_by_no.get(ono) or {} lines.append({ "vlNo": "—", "orderNo": ono, "product": _flex_product_name(store, o.get("productCode") or ""), "quantity": o.get("quantity"), "start": None, "end": None, "due": o.get("dueDate"), "priority": o.get("priority"), "stepCount": 0, "onTime": False, "status": "blocked", "steps": [], "blockReason": c.get("description") or c.get("conflictType"), }) scheduled_nos.add(ono) lines.sort(key=lambda x: (0 if x.get("status") == "scheduled" else 1, x.get("due") or "", x.get("orderNo") or "")) conflicts = [{ "type": c["conflictType"], "orderNo": c.get("orderNo"), "severity": c.get("severity"), "description": c.get("description"), "suggestion": c.get("suggestedSolution"), } for c in store.data["flexConflicts"] if c["versionId"] == result["versionId"]] if has_adoption_flow((store.data.get("planningContext") or {}).get("sourceProfile")): total_orders = int(result.get("orderCount") or 0) scheduled_orders = len(vls) blocked_orders = max(0, total_orders - scheduled_orders) trial_block = UIBlock(blockId=f"flex-sched-{result['versionId']}", type="flex-schedule", props={ "versionNo": result["versionNo"], "modeCn": "历史资料试排", "trialOnly": True, "productionReady": False, "solveStatus": solve_status, "assumptionCount": int(result.get("assumptionCount") or 0), "stats": {"vlCount": scheduled_orders, "woCount": result["woCount"], "makespan": result.get("makespan"), "onTimeCount": result.get("onTimeCount", 0), "conflictCount": result["conflictCount"], "utilization": round(float(result.get("avgUtilization") or 0) * 100), "orderCount": total_orders, "blockedOrderCount": blocked_orders}, "bottleneck": [], "lines": lines, "conflicts": conflicts, "downloadUrl": None, "filename": None, }) status_text = "试排暂未生成可执行方案" if not scheduled_orders else "已生成试排安排" assumption_count = int(result.get("assumptionCount") or 0) assumption_note = (f"其中 {assumption_count} 项按「试排假设」处理" "（未登记人员技能、未确认在制事实已逐条列出），" if assumption_count else "") return AgentReply( text=f"{status_text}：本次 {total_orders} 张订单，已安排 {scheduled_orders} 张，" f"待补资料 {blocked_orders} 张，生成 {result['woCount']} 个工单。" + assumption_note + "请核对下面的人员、在制和供料问题；已采用的主数据修改会保留。" "这是带来源标记的试排结果，尚未下发到车间。", blocks=[trial_block], commands=( [ViewportCommand(cmd="viewport.mode", params={"mode": "flex"}, issuedBy="LLM")] if scheduled_orders else [] ) + [ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")], ) block = UIBlock( blockId=f"flex-sched-{result['versionId']}", type="flex-schedule", props={ "versionNo": result["versionNo"], "modeCn": mode_cn, "stats": {"vlCount": result["vlCount"], "woCount": result["woCount"], "makespan": result.get("makespan"), "onTimeCount": result.get("onTimeCount", 0), "conflictCount": result["conflictCount"], "utilization": round(float(result.get("avgUtilization") or 0) * 100)}, "bottleneck": [{"name": b["operationName"], "count": b["poolEquipmentCount"]} for b in result.get("bottleneck", [])], "lines": lines, "conflicts": conflicts, "orderNo": intent.params.get("orderNo"), "downloadUrl": None, "filename": None, "emptyHint": None if lines else "本版没有排出虚拟产线：多半是订单缺工艺/设备。请先「分析一下这个项目」。", }) # 同步生成 Excel 工作计划表，排产结果块可直接下载 try: plan = build_report(store.data, "plan", order_no=intent.params.get("orderNo")) if plan.get("reportId") and plan.get("xlsxBytes"): saved = persist_report_xlsx(plan["reportId"], plan["xlsxBytes"], plan.get("filename")) block.props["downloadUrl"] = f"/api/reports/files/{plan['reportId']}" block.props["filename"] = saved["filename"] block.props["reportId"] = plan["reportId"] except Exception: pass deferred = result.get("deferredCount") or 0 focus_no = intent.params.get("orderNo") focus_hint = f"订单 {focus_no} · " if focus_no and order_ids else "" dl_hint = "点结果块「下载 Excel 工作计划表」可导出完整工序计划。" if block.props.get("downloadUrl") else "可说「生成排产方案报告」下载。" guide_block = None if solve_status in {"BLOCKED", "REJECTED", "INFEASIBLE", "ERROR"}: planning_summary = ((result.get("planning") or {}).get("summary") or {}) blocker_counts = planning_summary.get("blockerCounts") or {} blocker_text = "、".join( f"{code}x{count}" for code, count in sorted(blocker_counts.items())[:5] ) from server.aps_domain.guidance import scheduling_data_guide data_guide = scheduling_data_guide(store.data) if data_guide.get("steps"): guide_block = UIBlock(blockId="guidance-data-missing", type="guidance", props=data_guide) text = ( f"闭环排产被阻断 ⚠️ {focus_hint}版本 {result['versionNo']}：" f"制造需求 {int(result.get('demandCount') or 0)} 项，" f"准入 {int(result.get('admittedDemandCount') or 0)} 项，" f"未排 {int(result.get('unscheduledDemandCount') or 0)} 项；" f"未生成可发布生产工单或 MES 下发对象。" + (f"主要阻断：{blocker_text}。" if blocker_text else "") + dl_hint + ("\n\n别担心，我把还缺的东西列在下面，你按顺序补就行。" if guide_block else "") ) else: text = ( f"柔性排产完成。{focus_hint}版本 {result['versionNo']}（{mode_cn}·{win_cn}）：" f"{result['vlCount']} 条虚拟产线 / {result['woCount']} 个工单，冲突 {result['conflictCount']} 项" + (f"，窗外延期 {deferred}" if deferred else "") + f"。{dl_hint}" ) return AgentReply(text=text, blocks=[block] + ([guide_block] if guide_block else []), commands=[ # 联动：切柔性甘特 + 重拉世界数据 ViewportCommand(cmd="viewport.mode", params={"mode": "flex"}, issuedBy="LLM"), ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")]) def _flex_reschedule(store: WorldStore, intent: IntentResult, session_id: str, actor: str) -> AgentReply: """分级重排（DY-01）：L2/L3/L4 一律出 P2 确认卡。""" from server.aps_domain.flex import preview_reschedule from server.agent_core import harness level = str(intent.params.get("level") or "L2").upper() title, lines = preview_reschedule(store, level) params = {"level": level, "sortMode": intent.params.get("sortMode")} block = harness.stage_confirmation(session_id, "flex.reschedule", params, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="flex.reschedule.stage", target={"type": "FLEX", "id": level}, power="P2", rationale={"confirmId": block.props["confirmId"], "level": level}) store.save() return AgentReply(text=f"{title} 属于 P2，需要你确认后执行。", blocks=[block]) def _flex_swap(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply: """L1 局部换机：池内备机接手。""" from server.aps_domain.flex import local_swap_equipment code = intent.params.get("equipmentCode") wo_id = intent.params.get("woId") try: r = local_swap_equipment( store, equipment_code=code, wo_id=int(wo_id) if wo_id else None, target_code=intent.params.get("targetCode"), mark_fault=bool(intent.params.get("markFault", True)), actor=actor) except ValueError as exc: return AgentReply(text=str(exc)) lines = [f"{s['orderNo']} {s['operationCode']}：{s['from']} → {s['to']}" for s in r["swapped"][:5]] more = f" 等 {len(r['swapped'])} 条" if len(r["swapped"]) > 5 else "" fail_n = len(r.get("failed") or []) text = (f"L1 局部换机完成 ✅ {len(r['swapped'])} 条工单改挂备机" + (f"；{fail_n} 条无备机" if fail_n else "") + (f"；设备 {r['faulted']} 已标故障" if r.get("faulted") else "") + ("：" + "；".join(lines) + more if lines else "。")) return AgentReply(text=text, commands=[ ViewportCommand(cmd="viewport.mode", params={"mode": "flex"}, issuedBy="LLM"), ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM"), ]) def _flex_capacity(store: WorldStore) -> AgentReply: """瓶颈产能法：短文案 + flex-capacity 块（各池日产能条形 + 瓶颈标注）。""" from server.aps_domain.flex import capacity_analysis cap = capacity_analysis(store.data) if not cap["pools"]: return AgentReply(text="尚无可用的柔性产能池数据（无在运设备）。") max_daily = max((p["dailyCapacity"] for p in cap["pools"]), default=1) or 1 pools = [{ "name": p["operationName"], "equipmentCount": p["equipmentCount"], "dailyCapacity": p["dailyCapacity"], "workDays": p["workDays"], "isBottleneck": p["isBottleneck"], "pct": round(p["dailyCapacity"] / max_daily * 100), # 相对最大池的产能占比（条形宽度） } for p in cap["pools"]] bp = cap["bottleneckPool"] block = UIBlock(blockId="flex-cap", type="flex-capacity", props={"bottleneckName": bp["operationName"], "bottleneckDaily": bp["dailyCapacity"], "pools": pools}) text = (f"瓶颈产能评估完成：限制性瓶颈是【{bp['operationName']}】，" f"全厂节拍上界≈{bp['dailyCapacity']}件/天（按能力池实时算）。") return AgentReply(text=text, blocks=[block], commands=[ # EX-10：切产能池仪表盘 + 刷新世界 ViewportCommand(cmd="viewport.mode", params={"mode": "pool"}, issuedBy="LLM"), ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")]) def _flex_simulate_due(store: WorldStore, intent: IntentResult) -> AgentReply: """交期承诺模拟：短文案 + flex-due 块（乐观/预计/悲观 + 缺口）。""" from server.aps_domain.flex import simulate_due code = str(intent.params.get("productCode") or "") qty = int(intent.params.get("quantity") or 0) if not code or qty <= 0: return AgentReply(text="请告诉我产品与数量，例如：「PDU 500套什么时候能交」或「高压线束 800件多久能做完」。") r = simulate_due(store, code, qty, sort_mode=intent.params.get("sortMode")) block = UIBlock(blockId=f"flex-due-{code}", type="flex-due", props={"productName": _flex_product_name(store, code), "quantity": qty, "optimistic": r.get("optimisticFinish"), "expected": r.get("expectedFinish"), "pessimistic": r.get("pessimisticFinish"), "gaps": r.get("gaps") or [], "bottleneck": [{"name": b["operationName"], "count": b["poolEquipmentCount"]} for b in r.get("bottleneck", [])]}) gaps_txt = "；".join(g["message"] for g in (r.get("gaps") or [])[:2]) text = (f"交期承诺模拟完成：{_flex_product_name(store, code)} × {qty}" f"（乐观 {r.get('optimisticFinish') or '—'} / 预计 {r.get('expectedFinish') or '—'} / " f"悲观 {r.get('pessimisticFinish') or '—'}）。{gaps_txt}") return AgentReply(text=text, blocks=[block]) def _flex_compare(store: WorldStore) -> AgentReply: from server.aps_domain.analytics import matrix_from_flex_rows from server.aps_domain.flex import compare_sort_modes cmp = compare_sort_modes(store, compress_due=False) # 按真实交期试排 table = matrix_from_flex_rows( cmp["rows"], hint=cmp.get("hint"), data_quality=cmp.get("dataQuality"), due_basis=cmp.get("dueBasis")) block = UIBlock(blockId="flex-compare", type="flex-compare", props={"rows": cmp["rows"], "hint": cmp["hint"], "compressedDue": cmp["compressedDue"], "dueBasis": cmp.get("dueBasis"), "dataQuality": cmp.get("dataQuality"), "productionReady": table["productionReady"], "recommendationStatus": table["recommendationStatus"], "table": table}) recommendation_id = table.get("recommendationId") best = next((row for row in cmp["rows"] if row.get("sortMode") == recommendation_id), None) if best is None: text = ( "五模式对比完成（已压缩交期制造压力）。当前没有可推荐策略。" f"{table['recommendationReason']}已打开对比表视口。" ) elif table["recommendationStatus"] == "TRIAL_ONLY": text = ( "五模式对比完成（已压缩交期制造压力）。当前仅供试排参考，试排较优：" f"【{best['label']}】延迟 {best['totalTardiness']}h、" f"准时 {best['onTimeCount']}/{best['orderCount']}。" f"{table['recommendationReason']}；不能据此下达生产计划。已打开对比表视口。" ) else: text = ( "五模式对比完成（已压缩交期制造压力）。当前推荐：" f"【{best['label']}】延迟 {best['totalTardiness']}h、" f"准时 {best['onTimeCount']}/{best['orderCount']}。" "已打开对比表视口。" ) return AgentReply( text=text, blocks=[block], commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "compare"}, issuedBy="LLM")]) def _flex_rush(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply: from server.aps_domain.flex import insert_rush_order code = str(intent.params.get("productCode") or "HV-HARNESS") qty = int(intent.params.get("quantity") or 50) try: r = insert_rush_order(store, code, qty, due_date=intent.params.get("dueDate"), priority=int(intent.params.get("priority") or 1), actor=actor) except ValueError as exc: return AgentReply(text=str(exc)) o = r["order"] s = r["schedule"] text = (f"急单 {o['orderNo']} 已插入（{_flex_product_name(store, code)} × {qty}，交期 {o['dueDate']}），" f"瓶颈锚重排完成：版本 {s['versionNo']}，虚拟产线 {s['vlCount']}，" f"急单完工 {r.get('plannedEnd') or '—'}，冲突 {s['conflictCount']}。") return AgentReply(text=text, commands=[ ViewportCommand(cmd="viewport.mode", params={"mode": "flex"}, issuedBy="LLM"), ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM"), ]) def _flex_fault(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply: from server.aps_domain.flex import apply_equipment_fault code = str(intent.params.get("equipmentCode") or "PRESS-01") status = str(intent.params.get("status") or "MAINTENANCE").upper() try: r = apply_equipment_fault(store, code, status=status, reschedule=True, actor=actor) except ValueError as exc: return AgentReply(text=str(exc)) eq = r["equipment"] bn_before = r.get("capacityBefore") or {} bn_after = r.get("capacityAfter") or {} no_cap = [c for c in r.get("conflicts") or [] if c.get("conflictType") == "NO_CAPABILITY"] level = r.get("responseLevel") text = (f"设备 {eq['code']}（{eq['name']}）{eq['beforeStatus']} → {eq['afterStatus']}。" f"瓶颈日产能 {bn_before.get('dailyCapacity', '—')} → {bn_after.get('dailyCapacity', '—')} 件/天。") if level == "L1": n = len((r.get("swap") or {}).get("swapped") or []) text += f" 已按 L1 局部换机处理（{n} 条工单改挂备机，他单不动）。" elif no_cap: text += f" 出现 {len(no_cap)} 条无能力设备冲突（单点瓶颈停摆风险）。" elif eq["afterStatus"] == "MAINTENANCE" and "PRESS" in eq["code"]: text += " 压接池仍有冗余；L1 不可用时已缩池重排。" s = r.get("schedule") or {} if s: text += f" 已重排版本 {s.get('versionNo')}，冲突 {s.get('conflictCount', 0)}。" return AgentReply(text=text, commands=[ ViewportCommand(cmd="viewport.mode", params={"mode": "pool" if level != "L1" else "flex"}, issuedBy="LLM"), ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM"), ]) def _conflict_list(store: WorldStore, intent: IntentResult) -> AgentReply: from server.aps_domain.conflicts import list_conflict_center from server.contracts import UIBlock scope = str(intent.params.get("scope") or "flex") data = list_conflict_center(store.data, scope=scope) if not data["rows"]: return AgentReply(text="当前最新版本无未解决冲突 ✅") lines = [] for r in data["rows"][:8]: top = (r["fixes"][0]["label"] if r.get("fixes") else "—") lines.append(f"#{r['id']} [{r['severity']}] {r['conflictType']}：{r['description']} → {top}") more = f"\n…共 {data['total']} 条" if data["total"] > 8 else "" block = UIBlock( blockId="conflict-center", type="conflict-center", props={"total": data["total"], "critical": data["critical"], "major": data["major"], "rows": data["rows"][:20]}) return AgentReply( text=(f"冲突中心：共 {data['total']} 条（危急 {data['critical']} / 重大 {data['major']}）。\n" + "\n".join(lines) + more + "\n可在柔性工作台「冲突」页一键修复。"), blocks=[block], commands=[ViewportCommand(cmd="viewport.highlight", params={"what": "conflict"}, issuedBy="LLM")], ) def _conflict_resolve(store: WorldStore, intent: IntentResult, session_id: str, actor: str) -> AgentReply: from server.aps_domain.conflicts import apply_conflict_fix, list_conflict_center from server.contracts import UIBlock cid = intent.params.get("conflictId") action = intent.params.get("action") if not cid: data = list_conflict_center(store.data, scope="flex") if not data["rows"]: return AgentReply(text="没有可修复的冲突。") row = data["rows"][0] fixes = "；".join(f["label"] for f in row["fixes"][:3]) return AgentReply(text=f"建议先处理 #{row['id']}：{row['description']}\n可选：{fixes}") try: r = apply_conflict_fix(store, int(cid), str(action or "note"), params=intent.params.get("fixParams") or {}, actor=actor, session_id=session_id) except (ValueError, PermissionError) as exc: return AgentReply(text=str(exc)) blocks = [] if r.get("block"): blocks.append(UIBlock.model_validate(r["block"])) cmds = [ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")] if r.get("applied"): cmds.insert(0, ViewportCommand(cmd="viewport.mode", params={"mode": "flex"}, issuedBy="LLM")) return AgentReply(text=r.get("message") or "已处理", blocks=blocks, commands=cmds) def _ref_failure_text(exc: RefResolutionError) -> str: """引用解析失败的显式中文说明（fail closed：不吞掉、不假装命中）。""" if exc.kind == "knowledge": if "not-wired" in (exc.reason or ""): return ("「@知识」引用解析失败：知识库检索暂未接线，无法解析该引用，" "未编造任何知识内容。请联系管理员确认知识库可用后再试。") return (f"「@知识」引用解析失败：知识库未命中《{exc.target_id}》，未编造任何内容。" "可回复「知识库」查看资产清单，或核对资产 ID/标题关键词后重试。") label = {"session": "会话", "scenario": "方案", "version": "版本", "report": "报告"}.get(exc.kind, exc.kind) return (f"「@{label}」引用解析失败：目标《{exc.target_id}》不存在或无权访问" f"（{exc.reason}），本轮处理已停止，请核对引用对象。") def _resolve_refs_context(store: WorldStore, intent: IntentResult, actor: str) -> AgentReply | None: """对话链路入口的引用解析/注入（矩阵 63 行剩余项：@知识 + 接线到对话链路）。 - 无引用：返回 None，原意图处理不受影响； - 命中：解析结果注入 intent.params['_refs']（本轮上下文），并写 SESSION 审计（evidenceRefs 带 session:/scenario:/version:/report:/knowledge: 引用， 复用 round-15 trace/evidence 体系）； - 失败（@知识 未命中/未接线、引用对象删除/无权访问）：写 FAILED 审计并 返回显式中文说明（fail closed，绝不含混通过或假装命中）。 """ from server.agent_core.session_ref import RefResolutionError, parse_refs, resolve_all user_text = str((intent.params or {}).get("_rawUserText") or (intent.params or {}).get("query") or "") refs = parse_refs(user_text) if not refs: return None kb = get_knowledge() kb.ensure_seed_assets() def _knowledge_lookup(target: str) -> dict[str, Any] | None: q = target.strip() return kb.get(q) or kb.find_by_title(q) # assetId 精确 → 标题关键词 try: resolved, evidence_refs = resolve_all(store.data, refs, knowledge_lookup=_knowledge_lookup) except RefResolutionError as exc: write_audit(store.data, store.next_id, actor=actor, category="SESSION", action="session.ref.resolve", target={"type": "REF", "id": f"{exc.kind}:{exc.target_id}"}, power="P0", rationale={"reason": exc.reason, "refCount": len(refs)}, result="FAILED") store.save() return AgentReply(text=_ref_failure_text(exc)) intent.params["_refs"] = { "refs": [r.ref for r in resolved], "evidenceRefs": evidence_refs, "sources": [r.source for r in resolved], } write_audit(store.data, store.next_id, actor=actor, category="SESSION", action="session.ref.resolve", target={"type": "REF", "id": "|".join(evidence_refs)}, power="P0", rationale={"refCount": len(resolved), "refs": [r.ref for r in resolved]}, evidence_refs=evidence_refs) store.save() return None def _stage_pending_adoption(store: WorldStore, session_id: str, deep: dict[str, Any], actor: str) -> list[UIBlock]: """把只读分析出的可采用资料转成 P2 采用确认卡（未经批准不写世界）。 分析（P0）只出预览：这里把 Pi 结构化后的批次 / SQL 载荷 / MOM 来源文件原样 冻结进确认卡，人工批准后才由 execute_confirmed(action="import.commit") 写入； 三种来源（Excel/CSV 批次、SQL 数据包、MOM 主数据表）都没有可提交内容时不出卡。 整表替换类来源（SQL/MOM）额外冻结目标主数据指纹：审批窗口内世界漂移即拒绝执行。 """ from server.agent_core import harness from server.aps_domain.folder_pack import folder_schedule_world_fingerprint from server.aps_domain.importers import confirmation_for_import_commit pending = deep.get("pendingAdoption") or {} mode = str(pending.get("mode") or "") batches = [ batch for batch in (pending.get("batches") or []) if isinstance(batch, dict) and batch.get("okRows") ] raw_payload = pending.get("sqlPayload") sql_payload = copy.deepcopy(raw_payload) if isinstance(raw_payload, dict) and raw_payload else None mom_path = str(pending.get("applyPath") or "").strip() sql_stats = (sql_payload or {}).get("stats") or {} sql_row_count = _sql_payload_row_count(sql_payload) mom_business_rows = int(pending.get("businessRows") or 0) mom_counts = { str(key): int(value) for key, value in (pending.get("counts") or {}).items() if isinstance(value, (int, float)) and not isinstance(value, bool) and int(value) > 0 } filename = str(pending.get("filename") or "").strip() or "本次分析的资料" if batches: preview_batches = [{ "kind": batch.get("kind"), "sheet": batch.get("sheet") or batch.get("sourceFile") or filename, "okCount": len(batch.get("okRows") or []), "errorCount": int(batch.get("errorCount") or 0), } for batch in batches] total_ok = sum(item["okCount"] for item in preview_batches) total_errors = sum(item["errorCount"] for item in preview_batches) title, lines = confirmation_for_import_commit({ "filename": filename, "batches": preview_batches, "totalOk": total_ok, "totalErrors": total_errors, }) params: dict[str, Any] = {"filename": filename, "batches": batches} elif sql_payload is not None and sql_row_count > 0: stats = sql_stats title = f"采用分析资料 · {filename}" lines = [ "SQL 数据包只读解析完成：批准后整包写入项目主干（P2）", "订单 {} · 物料 {} · 工艺 {} · 设备 {} · BOM {}".format( stats.get("orders", 0), stats.get("materials", 0), stats.get("routing", 0), stats.get("equipment", 0), stats.get("bom", 0)), "批准后写入主干（P2）；执行前自动建档可回滚", ] params = {"filename": filename, "sqlPayload": sql_payload, "targetWorldFingerprint": folder_schedule_world_fingerprint(store.data)} elif mom_path and pending.get("adoptable") is True and mom_business_rows > 0 and mom_counts: detail = "、".join(f"{key} {value}" for key, value in mom_counts.items()) title = f"采用分析资料 · {filename}" lines = [ f"MOM 主数据表只读解析完成：{detail}", "批准后按这份文件替换项目资料（P2）", "批准后写入主干（P2）；执行前自动建档可回滚", ] params = {"filename": filename, "momPath": mom_path, "momSha256": pending.get("applySha256"), "targetWorldFingerprint": folder_schedule_world_fingerprint(store.data)} else: return [] # 采用获批后由 execute_confirmed 把这次分析写进租户知识库： # P0 分析只出预览，知识入库跟随 P2 采用一起发生。 params["analysisIngest"] = { "projectName": deep.get("projectName"), "summary": deep.get("summary") or {}, "sources": deep.get("sources") or [], "sourcePaths": (pending.get("sourcePaths") or [])[:8], "mode": pending.get("mode"), } block = harness.stage_confirmation( session_id, "import.commit", params, title=title, summary_lines=lines, evidence_refs=[ f"analysis-source:{path}" for path in (pending.get("sourcePaths") or [])[:8] ], ) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="import.commit.stage", target={"type": "IMPORT", "id": filename}, power="P2", rationale={"confirmId": block.props["confirmId"], "mode": mode, "batchCount": len(batches), "sqlPayload": sql_payload is not None, "momFrozen": bool(mom_path)}) store.save() return [block] def _analysis_summary_props( deep: dict[str, Any], *, has_adoption: bool, can_schedule: bool, ) -> dict[str, Any]: """Pi 侧展示摘要：首屏只给结论、关键数和下一步，详细内容留给折叠层。""" summary = deep.get("summary") or {} folder = deep.get("folder") or {} pending = deep.get("pendingAdoption") or {} issues: list[str] = [] seen: set[str] = set() def add_issue(message: Any) -> None: text = str(message or "").strip() if text and text not in seen: seen.add(text) issues.append(text) for item in folder.get("diagnostics") or []: if isinstance(item, dict) and item.get("severity") in ("blocking", "warning"): add_issue(item.get("message")) for line in deep.get("plan") or []: add_issue(line) metrics = [ {"key": "orders", "label": "订单", "value": int(summary.get("orders") or 0)}, {"key": "materials", "label": "产品和物料", "value": int(summary.get("materials") or 0)}, {"key": "routing", "label": "加工步骤", "value": int(summary.get("routings") or 0)}, {"key": "equipment", "label": "设备", "value": int(summary.get("equipment") or 0)}, ] mode = str(pending.get("mode") or "none") if has_adoption and mode != "none": status, status_label = "review", "待确认采用" headline = "资料已读取，确认采用后才会写入项目主数据。" next_step = None elif can_schedule: status, status_label = "ready", "可以排产" headline = "资料已通过检查，可以生成排产方案。" next_step = {"label": "开始排产", "command": "立即排产"} else: status, status_label = "blocked", "需要补充" headline = f"资料已读取，但有 {len(issues)} 项需要核对或补充。" next_step = {"label": "查看缺少资料", "command": "排产还缺什么"} return { "headline": headline, "status": status, "statusLabel": status_label, "summaryLine": ( f"订单 {metrics[0]['value']} · 产品和物料 {metrics[1]['value']} · " f"加工步骤 {metrics[2]['value']} · 设备 {metrics[3]['value']}" ), "metrics": metrics, "issues": issues[:3], "issueOverflow": max(0, len(issues) - 3), "nextStep": next_step, } async def handle_intent(store: WorldStore, session_id: str, intent: IntentResult, actor: str = "planner") -> AgentReply: """意图 → 回复 的总编排（gateway 调用；P0/P1 直通，P2 出卡）。""" name = intent.intent # 意图名 # 解析「这个订单」等指代 → 具体订单号 / flex orderIds if name in ("flex.schedule", "schedule.run", "plan.trace", "master.query", "report.generate"): intent = _resolve_order_reference(store, session_id, intent) if intent.params.get("_missingFocus") and name != "master.query": return AgentReply( text="刚才没有锁定具体订单。请先说「查询订单 102285668」，" "或直接说「给订单 102285668 排产」。") if intent.params.get("_missingFocus") and name == "master.query" and intent.params.get("entity") == "mrp": return AgentReply( text="刚才没有锁定具体订单。请先说「查询订单 102285668」，" "或直接问「订单 102285668 有委外吗」。") # ---- 结构化引用（矩阵 63 行）：@会话/@方案/@版本/@报告/@知识 ---- # 对话链路入口统一解析引用并注入证据链；未命中/未接线显式中文说明（fail closed）。 ref_fail = _resolve_refs_context(store, intent, actor) if ref_fail is not None: return ref_fail # ---- 工程图纸：目录解析 / SVG 预览 / 主数据候选 P2 暂存 ---- if name == "data.analyze" and intent.params.get("drawingAction") in ("analyze", "preview", "stage"): from server.aps_domain.folder_pack import analyze_work_dir report = analyze_work_dir(store.data, session_id) drawings = report.get("drawings") or [] requested = str(intent.params.get("path") or "").strip().lower() drawing = next((row for row in drawings if not requested or str(row.get("path") or "").lower() == requested or str(row.get("name") or "").lower() in requested), None) if not drawing: return AgentReply(text=(report.get("error") or "当前工程目录没有识别到 DXF 图纸。请把图纸放入项目工程目录后说“解析图纸”。")) if drawing.get("status") != "ok": return AgentReply(text=f"图纸 {drawing.get('name')} 解析失败：{drawing.get('issue')}。表格排产未受影响。") drawing_id = str(drawing.get("id") or "") asset = next((row for row in (store.data.get("drawingAssets") or []) if row.get("id") == drawing_id), {}) change_set = next((row for row in (store.data.get("drawingCandidates") or []) if row.get("drawingAssetId") == drawing_id), {}) props = { "drawingId": drawing_id, "asset": asset, "drawing": drawing.get("drawing") or {}, "candidateCounts": drawing.get("candidateCounts") or {}, "candidates": change_set, "previewUrl": f"/api/drawings/{drawing_id}/preview", "stageUrl": f"/api/drawings/{drawing_id}/stage", "reviewRequired": True, "writesMasterData": False, } drawing_block = UIBlock( blockId=f"drawing-{drawing_id}", type="drawing-analysis", props=props, actions=[], evidenceRefs=[str(asset.get("sha256") or "")]) if intent.params.get("drawingAction") == "stage": selected_candidates = [*(change_set.get("materials") or []), *(change_set.get("bomReferences") or []), *(change_set.get("routingOperations") or [])] payload = {"drawingId": drawing_id, "changeSetId": change_set.get("changeSetId"), "sourceSha256": (asset.get("parsed") or {}).get("asset", {}).get("sha256") or asset.get("sha256"), "selectedCandidates": selected_candidates, "candidateIds": [row.get("candidateId") for row in selected_candidates], "executionSupported": True} lines = [ f"图纸：{drawing.get('name')}", f"物料候选 {len(payload.get('materials') or [])} 项", f"BOM 引用 {len(payload.get('bomReferences') or [])} 项（数量未解析，不得正式写入）", f"工艺候选 {len(payload.get('routingOperations') or [])} 项（工时与资源待补全）", "当前只生成 P2 工程确认，不直接写正式主数据。", ] confirm_block = harness.stage_confirmation( session_id, "drawing.master.apply", payload, title="确认图纸主数据候选", summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="drawing.master.stage", target={"type": "DRAWING", "id": drawing_id}, power="P2", rationale={"confirmId": confirm_block.props.get("confirmId"), "executionSupported": True}) store.save() return AgentReply( text="已把图纸解析结果整理为物料、BOM 引用和工艺候选，并生成 P2 工程确认。当前不会写正式主数据；批准执行器仍列为待开发。", blocks=[drawing_block, confirm_block]) store.save() verb = "预览" if name == "drawing.preview" else "解析" return AgentReply( text=f"图纸 {drawing.get('name')} 已完成{verb}。候选数据需工程审核，未写入正式主数据。", blocks=[drawing_block]) # ---- 排产（P1） ---- if name == "schedule.run": return _run_schedule(store, intent, actor) # 试排并回执 # ---- 发布（P2 → 确认卡） ---- if name == "schedule.publish": return _stage_publish(store, session_id, actor, prefer_flex=True) # ---- 重置数据（P2 → 确认卡） ---- if name == "data.reset": return _stage_reset(store, session_id, actor) # ---- 新建/编辑订单（P2 → 确认卡；对话侧建单） ---- if name == "order.upsert": return _stage_order_upsert(store, session_id, intent, actor) # ---- 订单状态动作（P2 → 确认卡；对话与页面共用 action） ---- if name in ("order.cancel", "order.complete", "order.delete"): from server.aps_domain.orders import confirmation_for_order_action, find_order order_no = str(intent.params.get("orderNo") or "") order = find_order(store.data, order_no=order_no) if order is None: return AgentReply(text=f"没找到订单 {order_no}，可以在订单面板核对订单号。") title, lines = confirmation_for_order_action(store.data, name, {"id": order["id"]}) block = harness.stage_confirmation(session_id, name, {"id": order["id"]}, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action=name + ".stage", target={"type": "SALES_ORDER", "id": order["id"], "orderNo": order["orderNo"]}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 订单写入，需要你确认。", blocks=[block]) if name == "order.clear": from server.aps_domain.orders import confirmation_for_order_action try: title, lines = confirmation_for_order_action(store.data, "order.clear", {}) except ValueError as exc: return AgentReply(text=str(exc)) block = harness.stage_confirmation(session_id, "order.clear", {}, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="order.clear.stage", target={"type": "SALES_ORDER", "id": "ALL"}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 破坏性写入，需要你确认。", blocks=[block]) # ---- OR-03 订单池：提交（P1）/ 批准驳回（P2）/ 摘要（P0） ---- if name == "order.pool": from server.aps_domain.orders import list_orders, pool_summary sm = pool_summary(store.data) pending = [o for o in list_orders(store.data) if o["status"] == "SUBMITTED"] lines = [ f"订单池：全部 {sm['all']} · 可排产 {sm['schedulable']} · 待审 {sm['pending']} · " f"已批 {sm['approved']} · 变更 {sm['changed']} · 草稿/驳回 {sm['draft']}", ] if pending: lines.append("待审：" + "、".join(f"{o['orderNo']}({o['customerName']})" for o in pending[:8])) lines.append("可说「批量批准」或「批准订单 SOxxx」。") else: lines.append("当前无待审订单。新建订单默认为 DRAFT，需「提交审核」后再批准。") return AgentReply(text="\n".join(lines)) # ---- OR-04 紧急插单：快评（P1 沙盒）/ 采用（P2） ---- if name == "rush.evaluate": from server.aps_domain.rush import evaluate_rush, format_impact_text, impact_to_block p = dict(intent.params or {}) try: impact = harness.guard("rush.evaluate", p, lambda: evaluate_rush(store.data, p)) except ValueError as exc: return AgentReply(text=str(exc)) _LAST_RUSH_EVAL.clear() _LAST_RUSH_EVAL.update(impact) write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="rush.evaluate", target={"type": "SANDBOX", "id": impact["evalId"]}, power="P1", rationale={"affected": impact["affectedOrderCount"], "delayDelta": impact["delayDelta"], "conflictDelta": impact["conflictDelta"]}) store.save() text = format_impact_text(impact) blocks = [impact_to_block(impact)] lns = impact.get("lns") if lns: from server.aps_domain.lns import format_lns_text, lns_to_block text = text + "\n\n" + format_lns_text(lns) blocks.append(lns_to_block(lns)) return AgentReply( text=text, blocks=blocks, commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "due"}, issuedBy="LLM")]) if name == "rush.apply": p = dict(intent.params or {}) if not p.get("payload") and not p.get("orderNo") and not p.get("productCode") and not p.get("productId"): if not _LAST_RUSH_EVAL: return AgentReply(text="还没有插单快评结果。请先说「插单快评」或「紧急插单 200件控制器A」。") p = { "payload": _LAST_RUSH_EVAL.get("payload"), "strategy": _LAST_RUSH_EVAL.get("strategy"), "evalId": _LAST_RUSH_EVAL.get("evalId"), "lns": _LAST_RUSH_EVAL.get("lns"), } rush = p.get("payload") or p label = rush.get("orderNo") or f"{rush.get('customerName', '急单')}×{rush.get('quantity', '?')}" strategy_name = { "DELIVERY_FIRST": "交期优先", "CAPACITY_BALANCE": "产能均衡", "COMPREHENSIVE": "综合优化", }.get(str(p.get("strategy") or "DELIVERY_FIRST").upper(), "交期优先") if p.get("lns"): from server.aps_domain.lns import _planner_lns_reasons escalation_reasons = "；".join( _planner_lns_reasons(p["lns"].get("escalateReasons") or []) ) or "按当前生产安排无法只调整少数订单" else: escalation_reasons = "" block = harness.stage_confirmation( session_id, "rush.apply", p, title=f"采用紧急插单 · {label}", summary_lines=[ f"排产方式：{strategy_name}", "确认后会把这笔急单加入订单池，并生成一版待发布的调整方案", "现有已发布计划不改动；执行前会自动留存可回滚点", *([f"局部调整：{p['lns']['disturbance']['movedOrderCount']} 张订单需要顺延，" f"调整范围外的订单保持不变"] if p.get("lns") and p["lns"].get("status") == "LOCAL" else [f"整版调整：{escalation_reasons}"] if p.get("lns") else []), ], ) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="rush.apply.stage", target={"type": "SALES_ORDER", "id": rush.get("orderNo") or "NEW"}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text="采用这项插单方案前需要你确认，具体影响见下方确认卡。", blocks=[block]) # ---- OR-05 预测 / 长周期订单 ---- if name == "forecast.query": from server.aps_domain.forecast import forecast_summary, list_forecasts sm = forecast_summary(store.data) rows = list_forecasts(store.data) lines = [ f"预测订单：全部 {sm['all']} · 生效 {sm['active']} · 草稿 {sm['draft']} · " f"已转正 {sm['consumed']} · 取消 {sm['cancelled']}", ] for f in rows[:8]: lines.append( f"· {f['forecastNo']} {f.get('productName')}×{f.get('quantity')} " f"{f.get('periodStart')}~{f.get('periodEnd')} · {f.get('status')} · " f"置信{int((f.get('confidence') or 0)*100)}%" ) if sm["active"]: lines.append("可说「预测纳入试排」或「预测转正 FCxxx」。") return AgentReply(text="\n".join(lines)) if name in ("forecast.upsert", "forecast.delete", "forecast.convert"): from server.aps_domain.forecast import confirmation_for_forecast_action p = dict(intent.params or {}) try: title, lines = confirmation_for_forecast_action(store.data, name, p) except ValueError as exc: return AgentReply(text=str(exc)) block = harness.stage_confirmation(session_id, name, p, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action=f"{name}.stage", target={"type": "FORECAST_ORDER", "id": p.get("id") or p.get("forecastNo") or "NEW"}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block]) # ---- PL-01/PL-02 时间分桶 + 粗能力 ---- if name == "plan.buckets": from server.aps_domain.planning import build_plan_buckets p = dict(intent.params or {}) try: report = build_plan_buckets( store.data, mode=str(p.get("mode") or "HYBRID"), start_date=p.get("startDate"), horizon_days=int(p.get("horizonDays") or 90), include_forecast=bool(p.get("includeForecast", True)), capacity_mode=str(p.get("capacityMode") or "FINITE"), ) except ValueError as exc: return AgentReply(text=str(exc)) sm = report["summary"] cap_label = "无限" if report.get("capacityMode") == "INFINITE" else "有限" lines = [ f"时间分桶（{report['mode']} · {cap_label}产能）：桶 {sm['bucketCount']} · " f"超载 {sm['overCount']} · 预警 {sm['warnCount']} · 正常 {sm['okCount']}", f"日产能 {report['dailyCapacity']} 件 · 展望 {report['horizonDays']} 天 · " + ( f"峰值所需日产能 {sm.get('peakRequiredDaily')} · 有限参照超载桶 {sm.get('finiteOverCount')}" if report.get("capacityMode") == "INFINITE" else f"总负荷 {round(sm['overallLoad']*100)}%" ) + ("（含预测）" if report["includeForecast"] else "（仅确定订单）"), ] for b in report["buckets"]: show = b["status"] in ("OVER", "WARN") or b["demandQty"] > 0 or b.get("finiteStatus") == "OVER" if not show: continue if report.get("capacityMode") == "INFINITE": lines.append( f"· {b['label']}({b['kind']}) 需求{b['demandQty']} 需日产{b.get('requiredDaily')} " f"有限参照{round((b.get('finiteLoadRatio') or 0)*100)}% [{b.get('finiteStatus')}]" ) else: lines.append( f"· {b['label']}({b['kind']}) 需求{b['demandQty']}/能力{b['capacity']} " f"负荷{round(b['loadRatio']*100)}% [{b['status']}]" ) if len(lines) > 14: lines.append("…（完整表见右侧「分桶计划」视口）") break return AgentReply( text="\n".join(lines), commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "plan"}, issuedBy="LLM")], ) if name == "plan.rccp": from server.aps_domain.planning import build_rccp_compare p = dict(intent.params or {}) try: cmp_ = build_rccp_compare( store.data, mode=str(p.get("mode") or "HYBRID"), start_date=p.get("startDate"), horizon_days=int(p.get("horizonDays") or 90), include_forecast=bool(p.get("includeForecast", True)), ) except ValueError as exc: return AgentReply(text=str(exc)) d = cmp_["delta"] lines = [ f"RCCP 粗能力对照（{cmp_['mode']}）：{d['message']}", f"日产能缺口（峰值所需−当前）≈ {d['capacityGapDaily']} 件/天", ] for s in (cmp_.get("shortfallBuckets") or [])[:6]: lines.append( f"· {s['label']} 有限缺口 {s['finiteGap']} · 所需日产 {s['requiredDaily']} " f"· 负荷 {round((s.get('finiteLoadRatio') or 0)*100)}%" ) lines.append("可说「无限产能粗评」或在分桶视口切换有限/无限。") return AgentReply( text="\n".join(lines), commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "plan"}, issuedBy="LLM")], ) if name == "plan.feasibility": from server.aps_domain.planning import build_feasibility p = dict(intent.params or {}) try: report = build_feasibility( store.data, start_date=p.get("startDate"), horizon_days=int(p.get("horizonDays") or 90), include_forecast=bool(p.get("includeForecast", True)), ) except ValueError as exc: return AgentReply(text=str(exc)) sm = report["summary"] verdict = report["verdict"] lines = [ f"可行性结论：【{verdict}】 · 订单 {sm['all']} " f"（可行 {sm['feasible']} / 风险 {sm['atRisk']} / 不可行 {sm['infeasible']} / 过期 {sm['late']}）", f"日产能 {report['dailyCapacity']} · 超载桶 {sm['shortfallBucketCount']} · " f"分桶总负荷 {round((sm.get('overallLoad') or 0)*100)}%" + ("（含预测）" if report["includeForecast"] else "（仅确定订单）"), ] for g in (report.get("gaps") or [])[:5]: lines.append(f"· 缺口桶 {g['label']} 负荷{round((g.get('loadRatio') or 0)*100)}% 缺口{g.get('gap')}") risky = [o for o in report["orders"] if o["status"] in ("INFEASIBLE", "AT_RISK", "LATE")] for o in risky[:6]: reason = ("；".join(o.get("reasons") or []) or o["status"]) lines.append(f"· {o['ref']} 交期{o['dueDate']} slack={o['slack']} [{o['status']}] {reason}") if not risky: lines.append("当前累计粗能力下，已批准需求均可按期覆盖。") lines.append("完整表见右侧「分桶计划」→ 可行性。") return AgentReply( text="\n".join(lines), commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "plan"}, issuedBy="LLM")], ) if name == "plan.inventory": from server.aps_domain.inventory import build_inventory_projection p = dict(intent.params or {}) try: report = build_inventory_projection( store.data, material_code=p.get("materialCode"), material_type=p.get("materialType"), horizon_days=int(p.get("horizonDays") or 30), bucket=str(p.get("bucket") or "DAY"), include_forecast=bool(p.get("includeForecast", True)), start_date=p.get("startDate"), ) except ValueError as exc: return AgentReply(text=str(exc)) sm = report["summary"] lines = [ f"库存投影（{report['bucket']} · {report['horizonDays']}天）：" f"物料 {sm['materialCount']} · 断料 {sm['stockoutCount']} · " f"低于安全库存 {sm['belowSafetyCount']} · 正常 {sm['okCount']}", ] risky = [m for m in report["materials"] if m["alert"] != "OK"] for m in (risky or report["materials"])[:8]: lines.append( f"· {m['code']} {m['name']} 期初{m['openingStock']}→末{m['endingProjected']} " f"[{m['alert']}]" + (f" 首警{m['firstAlertDate']}" if m.get("firstAlertDate") else "") ) lines.append("完整曲线见右侧「分桶计划」→ 库存投影。") return AgentReply( text="\n".join(lines), commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "plan"}, issuedBy="LLM")], ) if name == "plan.leveling": from server.aps_domain.planning import build_leveling p = dict(intent.params or {}) try: report = build_leveling( store.data, mode=str(p.get("mode") or "WEEK"), start_date=p.get("startDate"), horizon_days=int(p.get("horizonDays") or 90), include_forecast=bool(p.get("includeForecast", True)), target_load=float(p.get("targetLoad") or 0.85), ) except ValueError as exc: return AgentReply(text=str(exc)) b, a, imp = report["before"], report["after"], report["improvement"] lines = [ f"产能削峰（{report['mode']} · 目标≤{int(report['targetLoad']*100)}%）：" f"峰值负荷 {round(b['peakLoad']*100)}%→{round(a['peakLoad']*100)}% · " f"超载桶 {b['overCount']}→{a['overCount']} · 建议挪动 {imp['moveCount']} 笔/{imp['movedQty']} 件", ] for m in report["moves"][:8]: arrow = "↑提前" if m["direction"] == "PULL_AHEAD" else "↓延后" lines.append( f"· {arrow} {m['quantity']}×{m.get('productCode') or '?'} " f"{m['fromLabel']}→{m['toLabel']}" ) if not report["moves"]: lines.append("当前分桶无明显超载，无需削峰。") lines.append("完整对照见右侧「分桶计划」→ 削峰。本切片不改交期。") return AgentReply( text="\n".join(lines), commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "plan"}, issuedBy="LLM")], ) if name == "plan.supply": from server.aps_domain.planning import build_supply_decisions p = dict(intent.params or {}) try: report = build_supply_decisions( store.data, mode=str(p.get("mode") or "WEEK"), start_date=p.get("startDate"), horizon_days=int(p.get("horizonDays") or 90), include_forecast=bool(p.get("includeForecast", True)), target_load=float(p.get("targetLoad") or 0.85), ) except ValueError as exc: return AgentReply(text=str(exc)) lines = [ f"产供决策【{report['verdict']}】：{report['message']}", f"削峰挪动 {report['leveling']['movedQty']} 件 · 残留缺口 {report['summary']['residualGap']} · " f"选项 {report['summary']['optionCount']} 条", ] for o in report["options"][:8]: lines.append( f"· [{o['label']}] {o['bucketLabel']} 覆盖{o['coverQty']} " f"成本指数{o['costIndex']} · {o['detail']}" ) if report["verdict"] == "BALANCED": lines.append("可继续试排或查看库存投影。") else: lines.append("完整方案见右侧「分桶计划」→ 产供。本切片不写主干。") return AgentReply( text="\n".join(lines), commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "plan"}, issuedBy="LLM")], ) if name == "order.submit": from server.aps_domain.orders import apply_order_submit, find_order p = dict(intent.params or {}) if not p.get("id") and not p.get("orderNo"): # 无单号：取最近一条可提交订单 cand = next((o for o in reversed(store.data.get("salesOrders", [])) if o.get("status") in ("DRAFT", "REJECTED", "CHANGED")), None) if not cand: return AgentReply(text="没有可提交审核的订单（需 DRAFT/REJECTED/CHANGED）。") p["id"] = cand["id"] try: applied = harness.guard("order.submit", p, lambda: apply_order_submit(store.data, p)) except ValueError as exc: return AgentReply(text=str(exc)) order = applied["order"] write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="order.submit", target={"type": "SALES_ORDER", "id": order["id"], "orderNo": order["orderNo"]}, power="P1", rationale={"from": applied.get("beforeStatus"), "to": "SUBMITTED"}) store.save() return AgentReply( text=(f"订单 {order['orderNo']} 已提交审核 ✅（{applied.get('beforeStatus')} → SUBMITTED）。" f"可说「批准订单 {order['orderNo']}」或在订单面板批量批准。"), commands=[ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")]) if name in ("order.approve", "order.reject"): from server.aps_domain.orders import confirmation_for_order_action, _resolve_order_ids payload = dict(intent.params or {}) if payload.get("orderIds") == "pending": payload["orderIds"] = "pending" elif payload.get("orderNo") and not payload.get("id"): pass try: # 预检可解析到订单 ids = _resolve_order_ids(store.data, payload) if not ids and payload.get("orderNo"): return AgentReply(text=f"订单 {payload['orderNo']} 不在待审状态（需 SUBMITTED/CHANGED）。") if not ids: return AgentReply(text="没有可审核的订单。可先「提交审核」或查看「订单池」。") payload["orderIds"] = ids title, lines = confirmation_for_order_action(store.data, name, payload) except ValueError as exc: return AgentReply(text=str(exc)) block = harness.stage_confirmation(session_id, name, payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action=name + ".stage", target={"type": "SALES_ORDER", "id": ids[0] if len(ids) == 1 else "BATCH"}, power="P2", rationale={"confirmId": block.props["confirmId"], "count": len(ids)}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block]) if name == "master.clear": from server.aps_domain.masterdata import confirmation_for_master_action scope = str(intent.params.get("scope") or "all") try: title, lines = confirmation_for_master_action(store.data, "master.clear", {"scope": scope}) except ValueError as exc: return AgentReply(text=str(exc)) block = harness.stage_confirmation(session_id, "master.clear", {"scope": scope}, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="master.clear.stage", target={"type": "MASTER_CLEAR", "id": scope}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 主数据写入，需要你确认。", blocks=[block]) # ---- 对话新建物料（P2） ---- if name == "master.material.upsert": from server.aps_domain.masterdata import confirmation_for_master_action payload = dict(intent.params or {}) missing = [label for key, label in (("code", "编码"), ("name", "名称")) if not payload.get(key)] if missing: return AgentReply(text=f"要新建物料还差：{'、'.join(missing)}。\n" "例如：「新建物料 编码 SCR-NEW 名称 不锈钢螺丝 类型 原料 库存 800」。") payload.setdefault("type", "RAW_MATERIAL") payload.setdefault("unit", "件") try: title, lines = confirmation_for_master_action(store.data, "master.material.upsert", payload) except ValueError as exc: return AgentReply(text=f"物料信息有误：{exc}") block = harness.stage_confirmation(session_id, "master.material.upsert", payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="master.material.upsert.stage", target={"type": "MATERIAL", "id": payload.get("code")}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 主数据写入，需要你确认。", blocks=[block]) if name == "changeover.query": from server.aps_domain.changeover import build_changeover_view view = build_changeover_view(store.data) lines = [ f"换型矩阵：{len(view['rows'])} 条显式规则 · 产品族 {len(view['families'])} 个", f"缺省跨族 {view['defaultCrossFamilyMin']} 分；同族 0 分", ] for r in view["rows"][:12]: lines.append(f"· {r['fromFamily']} → {r['toFamily']} = {r['setupMinutes']} 分" + (f"（{r['note']}）" if r.get("note") else "")) if not view["rows"]: lines.append("尚无显式规则，跨族走缺省。可说「设置换型 CTRL-STD 到 CTRL-HF 45分」。") lines.append("完整编辑见主数据 → 工艺模型 → 换型。") return AgentReply(text="\n".join(lines)) if name == "campaign.preview": from server.aps_domain.campaign import preview_campaigns p = dict(intent.params or {}) report = preview_campaigns( store.data, window_days=int(p.get("windowDays") or 7), include_forecast=bool(p.get("includeForecast", False)), ) s = report["summary"] lines = [ f"战役预览：输入 {s['inputOrders']} 单 → {s['campaignCount']} 场" f"（合并 {s['mergedCampaigns']} 场，省 {s['poSaved']} 个 PO）", f"交期窗口 {report['windowDays']} 天", ] for c in report["campaigns"][:10]: tag = "战役" if c["isCampaign"] else "单排" nos = "/".join(str(x) for x in (c.get("sourceOrderNos") or [])[:4]) lines.append( f"· [{tag}] {c['productCode']} ×{c['totalQty']} · {c['memberCount']} 单" f" · 交期 {c.get('earliestDue')}~{c.get('latestDue')} · {nos}" ) lines.append("确认后可说「战役合并试排」。") return AgentReply(text="\n".join(lines)) if name == "master.changeover.upsert": from server.aps_domain.masterdata import confirmation_for_master_action payload = dict(intent.params or {}) try: title, lines = confirmation_for_master_action(store.data, "master.changeover.upsert", payload) except ValueError as exc: return AgentReply(text=f"换型矩阵有误：{exc}") block = harness.stage_confirmation( session_id, "master.changeover.upsert", payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="master.changeover.upsert.stage", target={"type": "CHANGEOVER", "id": f"{payload.get('fromFamily')}→{payload.get('toFamily')}"}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 主数据写入，需要你确认。", blocks=[block]) # ---- 自然语言批量导入订单/物料（P2） ---- if name == "data.import": from server.aps_domain.intake import confirmation_for_import, parse_import # Pi 已经结构化；只接受显式 kind + rows，不再解析任何自然语言文本。 parsed = parse_import( intent.params.get("kind"), store.data, rows=intent.params.get("rows"), ) if not parsed["rows"]: err = "；".join(parsed.get("errors", [])[:3]) return AgentReply( text=("导入参数不完整或无效：" + (err or "请提供 kind 与非空 rows")) + "\n请提供结构化参数 kind=orders|materials 与 rows 数组；" "本工具不再从自然语言文本解析导入行。" ) title, lines = confirmation_for_import(parsed) payload = {"kind": parsed["kind"], "rows": parsed["rows"]} block = harness.stage_confirmation(session_id, "data.import", payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="data.import.stage", target={"type": "IMPORT", "id": parsed["kind"]}, power="P2", rationale={"confirmId": block.props["confirmId"], "count": len(parsed["rows"])}) store.save() return AgentReply(text=f"{title} 属于 P2 写入，需要你确认。", blocks=[block]) # ---- 订单分解（P1：MRP 建议草稿，不碰主数据与订单本体） ---- if name == "order.decompose": from server.aps_domain.mrp import decompose_orders, summarize_decomposition try: result = decompose_orders(store.data, store.next_id, intent.params.get("orderNo")) except ValueError as exc: return AgentReply(text=str(exc)) write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="order.decompose", target={"type": "MRP", "id": intent.params.get("orderNo") or "ALL"}, power="P1", rationale={"purchase": len(result["purchase"]), "outsource": len(result["outsource"])}) store.save() return AgentReply(text=summarize_decomposition(result)) # ---- 主数据/订单自然语言查询（P0 只读） ---- if name == "master.query": from server.agent_core.session_focus import set_focus from server.aps_domain.master_query import format_master_query, run_master_query code = intent.params.get("code") or intent.params.get("orderNo") # 「查这个订单」这类指代由 Pi 显式传 orderRef=last，这里只做确定性解析。 if str(intent.params.get("orderRef") or "").strip().lower() in ("last", "this", "focus"): from server.agent_core.session_focus import get_last_order_no code = get_last_order_no(session_id) or "" if not code: return AgentReply(text="本次会话还没有查到过订单，请先说出订单号或先查询订单。") result = run_master_query( store.data, entity=intent.params.get("entity"), code=code, aspect=intent.params.get("aspect"), ) # 记住刚查到的订单，供「把这个订单排产 / 有委外吗」指代 rows = result.get("rows") or [] if result.get("entity") == "order" and rows: set_focus(session_id, orderNo=rows[0].get("orderNo"), productCode=rows[0].get("productCode")) elif result.get("entity") == "mrp" and result.get("orderNo"): set_focus(session_id, orderNo=result.get("orderNo"), productCode=result.get("productCode")) elif result.get("productCode"): set_focus(session_id, productCode=result.get("productCode")) write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="master.query", target={"type": "MASTER", "id": result.get("entity") or "overview"}, power="P0", rationale={"entity": result.get("entity"), "code": code, "count": result.get("count")}) store.save() return AgentReply(text=format_master_query(result)) # ---- 计划追溯（P0：SO→主数据→分解→PO/WO→负荷/库存） ---- if name == "plan.trace": from server.aps_domain.trace import plan_trace, summarize_trace track = intent.params.get("track") try: result = plan_trace(store.data, intent.params.get("orderNo"), track=track) except ValueError as exc: return AgentReply(text=str(exc)) # both 形态：聊天只摘要；单轨带前 3 单 if result.get("track") == "both": block = UIBlock(blockId="plan-trace", type="plan-trace", props={"track": "both", "count": result["count"], "fixedCount": (result.get("fixed") or {}).get("count", 0), "flexCount": (result.get("flex") or {}).get("count", 0), "orders": ((result.get("flex") or {}).get("orders") or (result.get("fixed") or {}).get("orders") or [])[:2]}) else: preview = {**result, "orders": result["orders"][:3]} block = UIBlock(blockId="plan-trace", type="plan-trace", props=preview) hint = ("完整链路可在「柔性工作台 → 钉扎」或「订单 → 追溯」查看。" if (result.get("track") == "flex" or (intent.params.get("orderNo") or "").upper().startswith("FO-")) else "完整链路可在右侧「订单 → 追溯」查看。") return AgentReply(text=summarize_trace(result) + "\n" + hint, blocks=[block]) # ---- MRP 建议单下达（P2 → 确认卡：DRAFT 采购/委外转正式） ---- if name == "mrp.release": from server.aps_domain.mrp import confirmation_for_mrp_release order_no = str(intent.params.get("orderNo") or "") or None kind = str(intent.params.get("kind") or "all") try: title, lines = confirmation_for_mrp_release(store.data, order_no, kind) except ValueError as exc: return AgentReply(text=str(exc)) params = {"orderNo": order_no, "kind": kind} block = harness.stage_confirmation(session_id, "mrp.release", params, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="mrp.release.stage", target={"type": "MRP", "id": order_no or "ALL"}, power="P2", rationale={"confirmId": block.props["confirmId"], "kind": kind}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block]) # ---- 现场完整生产路线（P2：替换 flex*） ---- if name == "flex.site.load": from server.aps_domain.kangni_intake import build_flex_bundle, confirmation_for_site_load try: bundle = build_flex_bundle( intent.params.get("routePath") or None, intent.params.get("dataDir") or None, include_sibling_orders=intent.params.get("includeSiblings", False) is True, station_count=int(intent.params.get("stationCount") or 4), ) except (FileNotFoundError, ValueError) as exc: return AgentReply(text=f"无法加载现场路线：{exc}") meta = bundle.get("_meta") or {} title, lines = confirmation_for_site_load(meta) params = { "routePath": intent.params.get("routePath"), "dataDir": intent.params.get("dataDir"), "includeSiblings": intent.params.get("includeSiblings", False) is True, "stationCount": int(intent.params.get("stationCount") or 4), } block = harness.stage_confirmation(session_id, "flex.site.load", params, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="flex.site.load.stage", target={"type": "FLEX_SITE", "id": meta.get("primaryOrder")}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block]) # ---- 柔性排产（P1：能力池动态组虚拟产线，写草稿版本 §M5） ---- if name == "flex.schedule": return _run_flex(store, intent, actor) if name == "flex.reschedule": return _flex_reschedule(store, intent, session_id, actor) if name == "flex.swap": return _flex_swap(store, intent, actor) # ---- 瓶颈产能评估（P0：只读，瓶颈产能法） ---- if name == "flex.capacity": return _flex_capacity(store) # ---- 交期承诺模拟（P0：深拷贝沙盒试插单，不碰主干） ---- if name == "flex.simulate_due": return _flex_simulate_due(store, intent) if name == "flex.compare": return _flex_compare(store) if name == "flex.rush": return _flex_rush(store, intent, actor) if name == "flex.fault": return _flex_fault(store, intent, actor) if name == "conflict.list": return _conflict_list(store, intent) if name == "flex.conflict.resolve": return _conflict_resolve(store, intent, session_id, actor) # ---- SAP 集成（MD-05：状态 P0 / 入出站 P2 确认卡） ---- if name == "sap.status": from server.aps_domain.sap_sync import sap_connection_status st = sap_connection_status() return AgentReply( text=(f"SAP 连接：{'在线' if st.get('connected') else '离线'} · " f"{st.get('mode')} · {st.get('system')}/{st.get('plant')} · " f"待拉单 {st.get('openOrders')} · 已回写 {st.get('receiptCount')}。")) if name in ("sap.sync.inbound", "sap.sync.outbound"): from server.aps_domain.sap_sync import stage_sap_sync direction = "inbound" if name.endswith("inbound") else "outbound" r = stage_sap_sync(store, direction, session_id=session_id, actor=actor) return AgentReply(text=r["message"], blocks=[r["block"]]) # ---- MES 下发 / 报工（EX-05 / EX-09） ---- if name == "mes.status": from server.aps_domain.mes import list_execution, mes_connection_status st = mes_connection_status() ex = list_execution(store.data, "flex") return AgentReply( text=(f"MES：{'在线' if st.get('connected') else '离线'} · {st.get('system')} · " f"外部工单 {st.get('woCount')}（未完 {st.get('openCount')}）· " f"柔性已下发 {ex.get('total', 0)} / 完工 {ex.get('completed', 0)}。")) if name == "mes.dispatch": from server.aps_domain.mes import stage_dispatch r = stage_dispatch(store, track=intent.params.get("track") or "flex", session_id=session_id, actor=actor) if r.get("block"): return AgentReply(text=r["message"], blocks=[r["block"]]) return AgentReply(text=r.get("message") or "无法下发") if name == "mes.report": from server.aps_domain.mes import apply_report, list_execution track = intent.params.get("track") or "flex" wo_id = intent.params.get("woId") finish = bool(intent.params.get("finish") or intent.params.get("finishAll")) if intent.params.get("finishAll") or wo_id is None: ex = list_execution(store.data, track) targets = [r for r in ex.get("rows") or [] if r.get("status") != "COMPLETED" and (r.get("progressPct") or 0) < 100] if not targets: return AgentReply(text="没有已下发且未完工的工单。请先「下发MES」。") msgs = [] for row in targets[:30]: try: r = apply_report(store, int(row["woId"]), track=track, finish=True, actor=actor) msgs.append(r["message"]) except (ValueError, PermissionError) as exc: msgs.append(str(exc)) return AgentReply( text=f"批量报工完成 ✅ {len(msgs)} 条。\n" + "\n".join(msgs[:5]), commands=[ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")]) try: r = apply_report(store, int(wo_id), track=track, progress_pct=intent.params.get("progressPct"), finish=finish, actor=actor) except (ValueError, PermissionError) as exc: return AgentReply(text=str(exc)) return AgentReply(text=r["message"], commands=[ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")]) # ---- 方案对比（P1：Explore 沙盒，不碰主干 §5.1/§9.7） ---- if name == "scenario.compare": from server.aps_domain.analytics import matrix_from_fixed_cards text, block = compare_scenarios(store.data) # 沙盒并行试排三策略 block.props["table"] = matrix_from_fixed_cards( block.props.get("cards") or [], baseline=block.props.get("baseline")) write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="scenario.compare", target={"type": "SANDBOX", "id": block.blockId}, power="P1", rationale={"strategies": [c["strategy"] for c in block.props["cards"]]}) # 沙盒运行留痕 store.save() # 审计落盘（世界业务数据未变） return AgentReply( text=text + "\n已打开方案对比表视口。", blocks=[block], commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "compare"}, issuedBy="LLM")]) if name == "scenario.sensitivity": from server.aps_domain.sensitivity import run_sensitivity, sensitivity_as_block strategy = str((intent.params or {}).get("strategy") or "COMPREHENSIVE") report = run_sensitivity(store.data, strategy=strategy) block = sensitivity_as_block(report) write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="scenario.sensitivity", target={"type": "SANDBOX", "id": block.blockId}, power="P1", rationale={"strategy": strategy, "factors": [r["factorId"] for r in report["rows"]]}) store.save() top = report["rows"][0] if report["rows"] else None head = ( f"敏感性分析完成（策略 {strategy}）。基线延期 {report['baseline']['tardiness']} h。" ) if top: head += f"\n最敏感因子：{top['label']}（摆幅 {top['swing']}）。" head += f"\n{report.get('hint') or ''}" return AgentReply(text=head, blocks=[block]) # ---- 手动建档（P1：只新增快照 §4.3） ---- if name == "checkpoint.create": label = intent.params.get("label") or f"手动存档 {fmt_date(today0())}" # 缺省自动命名 meta = _create_checkpoint(store, label=label, reason="manual", conversation_note="用户口令建档") # 建成对快照 write_audit(store.data, store.next_id, actor=actor, category="WORLD_WRITE", action="checkpoint.create", target={"type": "CHECKPOINT", "id": meta["pairId"]}, power="P1", rationale={"label": label}) # 建档留痕 store.save() # 审计落盘 return AgentReply(text=f"已创建检查点【{label}】（ID: {meta['pairId']}）✅\n" f"随时可以说“回滚到检查点 {meta['pairId']}”回到此刻。", commands=[ViewportCommand(cmd="world.refresh", issuedBy="SYSTEM")]) # 刷新时间线导轨 # ---- 回滚（P2 → 确认卡 §4.3） ---- if name == "checkpoint.rollback": ckpt = get_checkpoints() # 快照仓 pair_id = intent.params.get("pairId") # 指定目标（时间线点击合成口令携带） pair = ckpt.get(pair_id) if pair_id else ckpt.latest() # 缺省=最近一个 if pair is None: # 无档可回 return AgentReply(text="当前没有可回滚的检查点。可先执行“建一个检查点”。") block = harness.stage_confirmation( # 回滚是 P2：出确认卡 session_id, "checkpoint.rollback", {"pairId": pair["pairId"]}, title=f"回滚到检查点【{pair['label']}】", summary_lines=[ # 影响面说明 f"目标时刻：{pair['createdAt']}（当时版本 {pair.get('versionNo') or '无'}）", "当前世界状态将被整体替换（对话与世界成对回滚 §4.2）", "回滚前会自动再存档一次，本次回滚可撤销", ]) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="checkpoint.rollback.stage", target={"type": "CHECKPOINT", "id": pair["pairId"]}, power="P2", rationale={"confirmId": block.props["confirmId"]}) # 出卡留痕 store.save() # 审计落盘 return AgentReply(text=f"回滚是 P2 写操作，需要你确认（目标：{pair['label']}）。", blocks=[block]) # ---- 视口命令族（P0：直接回传命令由前端执行） ---- if name.startswith("viewport."): replies = { # 各命令的回执文案 "viewport.mode": {"gantt": "已切换为【甘特图】。", "load": "已切换为【负荷热力图】。", "due": "已切换为【交期承诺看板】。", "pool": "已切换为【产能池】。", "flex": "已切换为【柔性甘特】。", "kpi": "已切换为【KPI 仪表盘】。", "compare": "已切换为【方案对比表】。", "util": "已切换为【资源利用率】。", "plan": "已切换为【分桶计划】。", }.get(intent.params.get("mode", ""), "已切换视图。"), "viewport.filter": "已应用过滤，右侧只显示匹配的工单。", "viewport.focus": f"已聚焦产线 {intent.params.get('lineCode', '')}，其他产线已隐藏。", "viewport.highlight": "已高亮" + ("【超期风险】(红)" if intent.params.get("what") == "overdue" else "【冲突】(黄)") + "，其余淡出。", "viewport.timescale": "已切换时间粒度。", "viewport.reset": "已重置视口：清除过滤、聚焦与高亮。", } return AgentReply(text=replies.get(name, "已执行。"), commands=[ # 回传结构化视口命令 ViewportCommand(cmd=name, params=intent.params, # 命令与槽位原样下发 target=intent.params.get("lineCode"), issuedBy="LLM")]) # ---- 知识库检索（P0：只读带出处 §8.2；可选 LLM 综合） ---- if name == "knowledge.query": mode = str(intent.params.get("mode") or "").strip().lower() if mode not in {"catalog", "search"}: return AgentReply( text="知识查询需要结构化参数 mode=catalog 或 mode=search；" "本工具不再从 query 文本推断查询类型。" ) from server.agent_core.providers import get_provider from server.knowledge.retrieval import hybrid_search kb = get_knowledge() kb.ensure_seed_assets() q = str(intent.params.get("query") or "").strip() title_hint = str(intent.params.get("assetTitle") or "").strip() if mode == "catalog": metas = kb.list_meta() by_kind: dict[str, list] = {} for m in metas: by_kind.setdefault(m.get("kind") or "other", []).append(m["title"]) lines = [f"知识库共 {len(metas)} 条资产："] kind_cn = {"sop": "SOP", "algorithm": "算法", "process": "工艺", "case": "案例", "report": "报告"} for k, titles in by_kind.items(): lines.append(f"【{kind_cn.get(k, k)}】{len(titles)} 条") for t in titles[:12]: lines.append(f" · {t}") if len(titles) > 12: lines.append(f" · …另有 {len(titles) - 12} 条") lines.append("可说「@知识:机加工工艺路线生成总则」或「导入知识文档」扩展语料。") write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="knowledge.query", target={"type": "KNOWLEDGE", "id": "catalog"}, power="P0", rationale={"query": q, "mode": mode, "count": len(metas)}) store.save() return AgentReply(text="\n".join(lines)) if not q and not title_hint: return AgentReply( text="知识检索需要结构化参数 query（关键词）或 assetTitle（资产标题）；" "本次未执行检索。" ) if title_hint: asset = kb.find_by_title(title_hint) hits = [{"assetId": asset["assetId"], "title": asset["title"], "kind": asset["kind"], "version": asset["version"], "score": 1.0, "snippet": asset["content"][:120], "content": asset["content"]}] if asset else [] else: hits = hybrid_search(kb.iter_search_units(), q, top_k=5) write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="knowledge.query", target={"type": "KNOWLEDGE", "id": hits[0]["assetId"] if hits else "miss"}, power="P0", rationale={"query": q, "assetTitle": title_hint or None, "mode": mode, "hits": [h["assetId"] for h in hits]}) store.save() if not hits: from server.aps_domain.guidance import build_guidance block = UIBlock(blockId="guidance-kb-miss", type="guidance", props=build_guidance(store.data, context="knowledge_miss")) return AgentReply( text="知识库里没有找到相关内容（不编造）。\n" "可以换个说法，或说「知识库」查看已有资产清单。", blocks=[block]) # 尝试 LLM 综合；失败则吐最相关原文 top = hits[0] text = None ctx_parts = [] for i, h in enumerate(hits[:4], 1): ctx_parts.append( f"[{i}] 《{h['title']}》{h['version']}" + (f" §{h['heading']}" if h.get("heading") else "") + f"\n{h['content'][:800]}" ) sys_prompt = ( "你是 APS 排产知识助手。只根据给定资料回答，禁止编造资料外的数字与规定。" "回答末尾列出引用的资料标题与版本。" ) user_prompt = f"问题：{q}\n\n资料：\n" + "\n\n".join(ctx_parts) try: text = await get_provider().chat_text(sys_prompt, user_prompt) except Exception: text = None if not text: text = (f"{top['content']}\n\n" f"—— 出处：【{top['title']}】{top['version']}（{top['kind']}）") else: cites = "；".join(f"【{h['title']}】{h['version']}" for h in hits[:4]) if "出处" not in text and "【" not in text: text = text.rstrip() + f"\n\n—— 出处：{cites}" block = UIBlock( blockId=f"evidence-{top['assetId']}", type="evidence", props={"hits": [{k: h.get(k) for k in ("assetId", "title", "kind", "version", "score", "snippet", "chunkId", "heading", "page")} for h in hits]}) return AgentReply(text=text, blocks=[block]) # ---- 知识文档导入（P2 确认卡） ---- if name == "knowledge.import": from server.knowledge.ingest import ( confirmation_for_knowledge_import, preview_ingest, ) path = intent.params.get("path") or intent.params.get("filePath") if not path: return AgentReply(text="请指定文件路径，例如：「导入知识文档 D:/docs/sop.md」") try: preview = preview_ingest( os.path.basename(str(path)), path=str(path), kind=str(intent.params.get("kind") or "sop"), title=intent.params.get("title"), ) except (FileNotFoundError, ValueError) as exc: return AgentReply(text=str(exc)) # 确认卡不塞全文 chunks，只留摘要；执行时再读 path title, lines = confirmation_for_knowledge_import(preview) slim = { "path": str(path), "filename": preview["filename"], "title": preview["title"], "kind": preview["kind"], "chunkCount": preview["chunkCount"], "sectionCount": preview["sectionCount"], } block = harness.stage_confirmation( session_id, "knowledge.import", slim, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="knowledge.import.stage", target={"type": "KNOWLEDGE", "id": preview["filename"]}, power="P2", rationale={"confirmId": block.props["confirmId"], "chunkCount": preview["chunkCount"]}) store.save() return AgentReply(text=f"{title} 属于 P2，需要你确认后入库。", blocks=[block]) # ---- 外部算法 Skill ---- if name == "skill.list": from server.agent_core.skills import get_skills rows = get_skills().list() if not rows: return AgentReply(text="尚未登记外部算法 skill。") lines = ["已接入算法 Skill："] for s in rows: flag = "启用" if s.get("enabled") else "停用" lines.append(f"· {s['skill_id']} 【{s['name']}】{flag} · {s.get('endpoint')} · 轨={s.get('track')}") lines.append("可说「用外部算法试排」或「skill 健康检查」。") return AgentReply(text="\n".join(lines)) if name == "skill.health": from server.agent_core.skills import get_skills health = get_skills().health(intent.params.get("skillId")) lines = ["Skill 健康检查："] for h in health: ok = "OK" if h.get("ok") else "FAIL" lines.append(f"· {h['skill_id']} [{ok}] {h.get('detail')} " f"{('· ' + str(h.get('latencyMs')) + 'ms') if h.get('latencyMs') is not None else ''}") return AgentReply(text="\n".join(lines)) # Pi owns the follow-up conversation; this tool only returns current facts. if name == "schedule.wizard": name = "readiness.query" # ---- 数据齐备度 / 工时维护（M-B：排产数据闭环） ---- if name == "readiness.query": from server.aps_domain.readiness import check_readiness, readiness_text, time_matrix from server.aps_domain.guidance import scheduling_data_guide report = harness.guard("readiness.query", {}, lambda: check_readiness(store.data)) matrix = time_matrix(store.data) block = UIBlock( blockId="readiness-report", type="readiness", props={"summary": report["summary"], "orders": report["orders"], "globalIssues": report["globalIssues"], "timeMatrix": [r for r in matrix if r["source"] in ("推断", "待维护")][:40]}, ) guide = scheduling_data_guide(store.data) blocks = [block] if guide.get("steps"): blocks.append(UIBlock(blockId="guidance-data-missing", type="guidance", props=guide)) return AgentReply(text=readiness_text(report), blocks=blocks) if name in ("data.analyze", "folder.analyze", "folder.schedule"): from server.aps_domain.planning_intake import profile_intake_reply, scheduling_entry_actions source_index = intent.params.get("sourceIndex") source_file = intent.params.get("sourceFile") or intent.params.get("filename") source_hint = ( f"{int(source_index)}." if isinstance(source_index, int) and not isinstance(source_index, bool) else Path(str(source_file)).name if source_file else "" ) try: profile_reply = profile_intake_reply( store, session_id, schedule_requested=name == "folder.schedule", query=source_hint, schedule_current=lambda: _run_flex(store, IntentResult( intent="flex.schedule", params={}, confidence=1.0, source="LLM"), actor), ) except (OSError, PermissionError, ValueError) as exc: return AgentReply(text=f"资料检查未完成：{exc}。请检查文件后重试。") if profile_reply is not None: return profile_reply if name == "data.analyze": from server.aps_domain.project_analyze import analyze_project_deep from server.aps_domain.guidance import scheduling_data_guide deep = harness.guard( "data.analyze", {}, lambda: analyze_project_deep( store.data, session_id, apply_sql=False, query="", input_path=str(intent.params.get("path") or "") or None, next_id=store.next_id, apply=False, ), ) store.save() can_schedule = bool(deep.get("canSchedule")) data_guide = scheduling_data_guide(store.data) block = UIBlock(blockId="project-analyze", type="project-analyze", props={ "projectName": deep.get("projectName"), "workDir": deep.get("workDir"), "sources": deep.get("sources") or [], "summary": deep.get("summary") or {}, "orders": deep.get("orders") or [], "materials": deep.get("materials") or [], "routings": deep.get("routings") or [], "equipment": deep.get("equipment") or [], "plan": deep.get("plan") or [], "kbActions": deep.get("kbActions") or [], "sqlPreview": deep.get("sqlPreview"), "canSchedule": can_schedule, "nextActions": scheduling_entry_actions( can_schedule=can_schedule, reason="请先补齐订单、工艺、工时、设备或班次等排产资料。", ), "applied": deep.get("applied") or {}, "owner": deep.get("owner"), "knowledgeIngest": deep.get("knowledgeIngest") or {}, }) adoption_cards = _stage_pending_adoption(store, session_id, deep, actor) block.props["analysisSummary"] = _analysis_summary_props( deep, has_adoption=bool(adoption_cards), can_schedule=can_schedule, ) blocks = [block, *adoption_cards] if not can_schedule and data_guide.get("steps"): blocks.append(UIBlock(blockId="guidance-data-missing", type="guidance", props=data_guide)) if adoption_cards: text = ("已读取并整理本项目的数据。请先确认采用本次资料，采用后才会写入项目主数据，" "再继续排产。") if not can_schedule: text += "当前资料还未通过排产检查，补齐缺项前不会开始排产。" elif can_schedule: text = "已读取并整理本项目的数据。当前资料已通过排产检查，但尚未生成方案。" else: text = "当前资料未通过排产检查，未生成方案。下面已按顺序列出需要补充的数据；补齐前不会开始排产。" return AgentReply(text=text, blocks=blocks) if name == "folder.analyze": from server.aps_domain.project_analyze import analyze_project_deep from server.aps_domain.guidance import scheduling_data_guide deep = harness.guard( "folder.analyze", {}, lambda: analyze_project_deep( store.data, session_id, apply_sql=False, query="", input_path=str(intent.params.get("path") or "") or None, next_id=store.next_id, apply=False, ), ) store.save() report = deep.get("folder") or {} can_schedule = bool(deep.get("canSchedule")) data_guide = scheduling_data_guide(store.data) block = UIBlock(blockId="folder-pack", type="folder-pack", props={ "projectName": report.get("projectName") or deep.get("projectName"), "workDir": report.get("workDir") or deep.get("workDir"), "files": report.get("files") or [], "coverage": report.get("coverage") or {}, "coverageDetail": report.get("coverageDetail") or [], "summary": deep.get("summary") or {}, "missing": report.get("missing") or [], "softMissing": report.get("softMissing") or [], "canSchedule": bool(report.get("canSchedule")), "nextActions": scheduling_entry_actions( can_schedule=bool(report.get("canSchedule")), reason="请先补齐订单、物料、工艺、工时或设备等排产资料。", ), "totalOk": report.get("totalOk") or 0, "totalErrors": report.get("totalErrors") or 0, "skippedRows": report.get("skippedRows") or 0, "error": report.get("error"), }) project_block = UIBlock(blockId="project-analyze", type="project-analyze", props={ "projectName": deep.get("projectName"), "workDir": deep.get("workDir"), "sources": deep.get("sources") or [], "summary": deep.get("summary") or {}, "orders": deep.get("orders") or [], "materials": deep.get("materials") or [], "routings": deep.get("routings") or [], "equipment": deep.get("equipment") or [], "plan": deep.get("plan") or [], "kbActions": deep.get("kbActions") or [], "sqlPreview": deep.get("sqlPreview"), "canSchedule": can_schedule, "nextActions": scheduling_entry_actions( can_schedule=can_schedule, reason="请先补齐订单、工艺、工时、设备或班次等排产资料。", ), "applied": deep.get("applied") or {}, "owner": deep.get("owner"), "knowledgeIngest": deep.get("knowledgeIngest") or {}, }) adoption_cards = _stage_pending_adoption(store, session_id, deep, actor) block.props["analysisSummary"] = _analysis_summary_props( deep, has_adoption=bool(adoption_cards), can_schedule=can_schedule, ) blocks = [block, project_block, *adoption_cards] if not can_schedule and data_guide.get("steps"): blocks.append(UIBlock(blockId="guidance-data-missing", type="guidance", props=data_guide)) return AgentReply( text=("已检查文件并整理项目数据。请先确认采用本次资料，采用后才会写入项目主数据，" "再继续排产。" if adoption_cards else ("已检查文件并整理项目数据。当前资料已通过排产检查，但尚未生成方案。" if can_schedule else "当前资料未通过排产检查，未生成方案。下面已按顺序列出需要补充的数据；补齐前不会开始排产。")), blocks=blocks, ) if name == "folder.schedule": from server.aps_domain.guidance import scheduling_data_guide from server.aps_domain.folder_pack import ( confirmation_for_folder_schedule, folder_schedule_payload_digest, prepare_folder_schedule, ) # P2 staging is read-only for the business world. Parsed rows and SQL payload # are frozen below and bound to source/world fingerprints. try: report = prepare_folder_schedule(store.data, session_id) except (OSError, PermissionError, ValueError) as exc: return AgentReply(text=f"工程目录分析失败：{exc}") kangni_trial = bool(report.get("kangniDetected") and report.get("trialReady")) kangni_meta = report.get("kangniMeta") or {} analyze_block = UIBlock(blockId="folder-pack", type="folder-pack", props={ "projectName": report.get("projectName"), "workDir": report.get("workDir"), "files": report.get("files") or [], "coverage": report.get("coverage") or {}, "coverageDetail": report.get("coverageDetail") or [], "missing": [] if kangni_trial else (report.get("missing") or []), "genericMissing": report.get("missing") or [], "softMissing": report.get("softMissing") or [], "canSchedule": bool(report.get("canSchedule") or kangni_trial), "nextActions": scheduling_entry_actions( can_schedule=bool(report.get("canSchedule") or kangni_trial), reason="请先补齐订单、物料、工艺、工时、设备或班次等排产资料。", ), "totalOk": report.get("totalOk") or 0, "totalErrors": report.get("totalErrors") or 0, "skippedRows": report.get("skippedRows") or 0, "error": report.get("error"), "sqlApplied": bool(report.get("sqlApplied")), "sqlStats": report.get("sqlStats") or {}, "kangniDetected": bool(report.get("kangniDetected")), "trialReady": bool(report.get("trialReady")), "productionReady": report.get("productionReady"), "kangniMeta": kangni_meta, }) data_guide = scheduling_data_guide(store.data) missing_blocks = [analyze_block] if data_guide.get("steps"): missing_blocks.append(UIBlock(blockId="guidance-data-missing", type="guidance", props=data_guide)) has_core = ( bool(store.data.get("flexOrders")) and bool(store.data.get("flexRoutings")) and any(e.get("status") == "RUNNING" for e in (store.data.get("flexEquipment") or [])) ) if not report.get("batches") and not report.get("sqlApplied") and not kangni_trial: return AgentReply( text="文件中没有可导入的有效记录，当前无法生成排产方案。" "下面已按顺序列出需要补充的数据；补齐前不会开始排产。", blocks=missing_blocks, ) if report.get("missing") and not has_core and not kangni_trial: return AgentReply( text="**当前无法排产：**缺少 " + "、".join(report["missing"]) + "。下面已按顺序列出补充方法；补齐前不会开始排产。", blocks=missing_blocks, ) title, lines = confirmation_for_folder_schedule(report) if kangni_trial: quality = kangni_meta.get("resourceQuality") or {} production_ready = bool(report.get("productionReady")) if production_ready: title = f"康尼真实数据生产落地试排 · {report.get('projectName') or ''}" ready_line = "当前生产可用性：productionReady=true（设备/模具映射已齐，无共享占位）" action_line = "确认后生成本地 trial 版本；正式 MES 仍需可信供应与独立 P3 确认" else: title = f"康尼真实数据本地试排 · {report.get('projectName') or ''}" ready_line = "设备和模具的实际对应关系还未确认，目前只能试排" action_line = "确认后生成试排方案，不会发布生产计划或下发到车间" lines = [ f"冻结工作簿：{len(report.get('sourceManifest') or [])} 份", f"订单 {kangni_meta.get('orderCount', 0)} · 工艺记录 {kangni_meta.get('routingRecordCount', 0)}", f"设备 {quality.get('equipmentCount', 0)} · 模具 {quality.get('moldCount', 0)}", ready_line, action_line, ] slim = [] if kangni_trial else [ {"kind": b["kind"], "sheet": b.get("sheet"), "okRows": b.get("okRows") or []} for b in report["batches"] if b.get("okRows") ] params = { "boundAction": "folder.schedule", "workDir": report.get("workDir"), "projectId": report.get("projectId"), "targetWorldKey": report.get("projectId"), "sessionId": session_id, "batches": slim, "sortMode": "BOTTLENECK", "sqlApplied": bool(report.get("sqlApplied")), "sqlPayload": report.get("sqlPayload"), "sourceManifestVersion": report.get("sourceManifestVersion"), "sourceManifest": report.get("sourceManifest") or [], "sourceManifestDigest": report.get("sourceManifestDigest"), "folderWorldFingerprint": report.get("folderWorldFingerprint"), "kangniDetected": bool(report.get("kangniDetected")), "trialReady": bool(report.get("trialReady")), "productionReady": report.get("productionReady"), "kangniPayload": report.get("kangniPayload"), "kangniMeta": kangni_meta, } checkpoint = _create_checkpoint( store, label="工程目录审批前基线", reason="stage:folder.schedule", conversation_note="待确认导入工程目录并试排", ) if not checkpoint: return AgentReply(text="工程目录确认卡创建失败：无法建立审批前快照。") params["beforeSnapshot"] = str(checkpoint["pairId"]) params["folderPayloadDigest"] = folder_schedule_payload_digest(params) evidence_refs = [ f"folder-source:{params['sourceManifestDigest']}", f"folder-world:{params['folderWorldFingerprint']}", f"folder-payload:{params['folderPayloadDigest']}", f"folder-snapshot:{params['beforeSnapshot']}", ] block = harness.stage_confirmation( session_id, "folder.schedule", params, title=title, summary_lines=lines, evidence_refs=evidence_refs, before_snapshot=str(checkpoint["pairId"]), ) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="folder.schedule.stage", target={"type": "FOLDER_PACK", "id": report.get("projectId") or "project"}, power="P2", rationale={"confirmId": block.props["confirmId"], "totalOk": report.get("totalOk"), "kangniDetected": kangni_trial, "trialOnly": True, "sourceManifestDigest": report.get("sourceManifestDigest")}) store.save() if kangni_trial: production_ready = bool(report.get("productionReady")) if production_ready: tail = ( "设备和模具对应关系已齐。" "确认后生成试排方案；正式下发前仍需核对供料并单独审批。" ) else: tail = ( "设备和模具对应关系仍需核对。" "确认后仅生成试排方案，不会下发到车间。" ) return AgentReply( text="数据检查已完成，正在等待你确认生成方案。" + tail, blocks=[block], ) return AgentReply( text="数据检查已完成，尚未生成方案。请确认下方内容，再保存数据并生成一版试排方案。", blocks=[block], ) if name in ("assistant.reply", "unknown"): # Pi 是唯一自然语言入口：这里只投递结构化请求，不再做本地意图/话术兜底。 from server.agent_core import fallback_lane params = dict(intent.params or {}) params["_piPrimary"] = True pi_intent = intent.model_copy(update={"params": params}) reply = await fallback_lane.propose_reply( store, session_id, pi_intent, actor=actor) if reply is not None: return reply return AgentReply(text="智能助手服务暂不可用，本次未执行任何操作。请稍后重试。") if name == "flex.time.update": pc = str(intent.params.get("productCode") or "") op_code = str(intent.params.get("operationCode") or "") raw_std_min = intent.params.get("stdMin") if not op_code or raw_std_min is None or isinstance(raw_std_min, bool): return AgentReply( text="工时更新需要结构化参数 operationCode（工序编码）和 stdMin（正数分钟）；" "本次不从 query 文本猜测工序或工时。" ) try: std_min = float(raw_std_min) except (TypeError, ValueError): return AgentReply( text="stdMin 必须是正数（分钟），本次未生成工时更新确认卡。" ) from math import isfinite if not isfinite(std_min) or std_min <= 0: return AgentReply( text="stdMin 必须是有限正数（分钟），本次未生成工时更新确认卡。" ) # 允许用工序名匹配（「部装」→ 所有名称含部装的工序编码） codes = [op_code] if not any(r.get("operationCode") == op_code for r in store.data.get("flexRoutings") or []): named = [o["code"] for o in store.data.get("flexOperations") or [] if op_code in (o.get("name") or "")] if named: codes = named staged_blocks = [] title = f"更新工时：{pc or '全部产品'} × {'、'.join(codes)} → {std_min} 分钟/件" block = harness.stage_confirmation( session_id, "flex.time.update", {"productCode": pc, "operationCode": codes[0] if len(codes) == 1 else "", "operationCodes": codes, "stdMin": float(std_min), "source": "实测"}, title=title, summary_lines=[f"匹配工序：{'、'.join(codes)}", f"单件工时 → {std_min} 分钟（来源标记为「实测」）", "确认后写入柔性工艺路线并同步数据库。"]) staged_blocks.append(block) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="flex.time.update.stage", target={"type": "ROUTING_TIME", "id": f"{pc or '*'}/{op_code}"}, power="P2", rationale={"confirmId": block.props["confirmId"], "stdMin": std_min}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，请确认。", blocks=staged_blocks) if name == "skill.register": payload = dict(intent.params or {}) if not payload.get("skill_id") or not payload.get("endpoint"): return AgentReply(text="登记 skill 需要 skill_id 与 endpoint，例如：" "「登记算法skill id=algo.x endpoint=http://127.0.0.1:8101 name=我的算法」") title = f"登记算法 Skill {payload.get('skill_id')}" lines = [f"ID：{payload.get('skill_id')}", f"端点：{payload.get('endpoint')}", f"名称：{payload.get('name') or payload.get('skill_id')}", f"轨道：{payload.get('track') or 'flex'}"] block = harness.stage_confirmation(session_id, "skill.register", payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="skill.register.stage", target={"type": "SKILL", "id": payload.get("skill_id")}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2，确认后写入配置。", blocks=[block]) if name == "skill.enable": sid = intent.params.get("skillId") or intent.params.get("skill_id") enabled = intent.params.get("enabled") if not sid or not isinstance(enabled, bool): return AgentReply( text="启停算法 Skill 需要结构化参数 skillId（字符串）和 enabled（布尔值 true/false）；" "不再从 query 文本推断启停动作。" ) sid = str(sid) title = f"{'启用' if enabled else '停用'}算法 Skill {sid}" block = harness.stage_confirmation( session_id, "skill.enable", {"skill_id": sid, "enabled": bool(enabled)}, title=title, summary_lines=[f"目标：{sid}", f"状态 → {'启用' if enabled else '停用'}"]) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="skill.enable.stage", target={"type": "SKILL", "id": sid}, power="P2", rationale={"confirmId": block.props["confirmId"], "enabled": enabled}) store.save() return AgentReply(text=f"{title} 属于 P2，请确认。", blocks=[block]) # ---- 报告生成（P1：产出文档并入知识库 §9.10） ---- if name == "report.generate": report_type = str(intent.params.get("reportType") or "daily") # 报告类型（缺省日报） order_no = str(intent.params.get("orderNo") or "").strip() or None report = build_report(store.data, report_type, order_no=order_no) # 冻结快照 → 模板生成 if not report["reportId"]: # 前置条件不满足（无版本/不足两版） return AgentReply(text=report["markdown"]) # 报告入知识库（§9.10 规则3：报告本身成为 RAG 语料） asset = get_knowledge().add(kind="report", title=report["title"], content=report["markdown"], tags=[report_type, "报告"]) download_url = None download_name = None fmt = "md" xlsx = report.get("xlsxBytes") if xlsx and report_type in ("plan", "schedule-plan", "flex-plan"): saved = persist_report_xlsx(report["reportId"], xlsx, report.get("filename")) download_url = f"/api/reports/files/{report['reportId']}" download_name = saved["filename"] fmt = "xlsx" write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="report.generate", target={"type": "REPORT", "id": report["reportId"]}, power="P1", rationale={"reportType": report_type, "assetId": asset["assetId"], "orderNo": order_no, "format": fmt}) store.save() block = UIBlock( blockId=f"report-{report['reportId']}", type="report", props={"reportId": report["reportId"], "title": report["title"], "markdown": report["markdown"], "reportType": report_type, "assetId": asset["assetId"], "format": fmt, "downloadUrl": download_url, "filename": download_name}) tip = ("点下方「下载 Excel 工作计划表」保存完整工序计划。" if fmt == "xlsx" else "可预览并下载 Markdown。") return AgentReply( text=f"已生成【{report['title']}】（数字取自冻结快照）。\n{tip}" f"报告已入知识库（资产 {asset['assetId']}）。", blocks=[block]) # ---- KPI 查询 / 仪表盘（P0，EX-08） ---- if name == "query.kpi": from server.aps_domain.analytics import build_kpi_dashboard dash = build_kpi_dashboard(store.data) lines = [f"· {c['label']}：{c['value']}" for c in dash.get("cards") or []] text = "运营 KPI 仪表盘：\n" + ("\n".join(lines) if lines else "暂无数据，请先排产。") block = UIBlock(blockId="kpi-dash", type="kpi-dashboard", props=dash) return AgentReply( text=text, blocks=[block], commands=[ViewportCommand(cmd="viewport.mode", params={"mode": "kpi"}, issuedBy="LLM")]) # ---- 主动引导（P0，AG-07） ---- if name == "guidance.next": from server.aps_domain.guidance import build_guidance, scheduling_data_guide data_guide = scheduling_data_guide(store.data) if data_guide.get("steps"): step_lines = [ f"{i}. {st['title']}：{st['detail']} {st['how']}" for i, st in enumerate(data_guide["steps"], start=1) ] return AgentReply( text=(data_guide.get("hint") or "要开排还差一些数据：\n") + "\n".join(step_lines), blocks=[UIBlock(blockId="guidance-data-missing", type="guidance", props=data_guide)]) guide = build_guidance(store.data) lines = [f"· {s['label']}（{s['reason']}）" for s in guide.get("suggestions") or []] sig = guide.get("signals") or [] head = "当前建议下一步：\n" if not sig else f"状态：{sig[0]['title']}。建议：\n" return AgentReply( text=head + ("\n".join(lines) if lines else "暂无特别建议，可说「帮助」。"), blocks=[UIBlock(blockId="guidance-next", type="guidance", props=guide)]) # ---- 排产参数查询 / 更新（OR-02） ---- if name == "params.query": from server.aps_domain.params import get_schedule_params p = get_schedule_params(store.data) lw = p["customerLevelWeights"] ow = p["weights"] text = ( "当前排产参数：\n" f"· 客户等级权重：VIP={lw['VIP']} / A={lw['A']} / B={lw['B']} / C={lw['C']}\n" f"· 目标权重：交期={ow['tardiness']} / 成本={ow['cost']} / 利用率={ow['utilization']} / 均衡={ow['balance']}\n" f"· 展望期 {p['planningHorizonDays']} 天\n" "可在设置 → 排产参数调整，或说「提高VIP权重到10」「恢复默认排产参数」。" ) return AgentReply(text=text) if name == "params.update": from server.aps_domain.params import confirmation_for_params_update raw = dict(intent.params or {}) payload: dict[str, Any] = {} if raw.get("resetDefaults"): payload["resetDefaults"] = True else: levels: dict[str, float] = {} mapping = {"vipWeight": "VIP", "aWeight": "A", "bWeight": "B", "cWeight": "C", "VIP": "VIP", "A": "A", "B": "B", "C": "C"} for src, dst in mapping.items(): if raw.get(src) is not None and src in ("vipWeight", "aWeight", "bWeight", "cWeight", "VIP", "A", "B", "C"): try: levels[dst] = float(raw[src]) except (TypeError, ValueError): return AgentReply(text=f"{dst} 权重必须是数字。") if raw.get("customerLevelWeights"): payload["customerLevelWeights"] = raw["customerLevelWeights"] elif levels: payload["customerLevelWeights"] = levels if raw.get("weights"): payload["weights"] = raw["weights"] if raw.get("planningHorizonDays") is not None: payload["planningHorizonDays"] = raw["planningHorizonDays"] try: title, lines = confirmation_for_params_update(store.data, payload) except ValueError as exc: return AgentReply(text=f"排产参数有误：{exc}") block = harness.stage_confirmation(session_id, "params.update", payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="params.update.stage", target={"type": "SCHEDULE_PARAMS", "id": "scheduleParams"}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block]) # ---- 约束配置（SC-04） ---- if name == "constraint.profile.query": from server.aps_domain.constraints import get_constraint_profile profile = get_constraint_profile(store.data) lines = [] for c in profile["constraints"]: if not c.get("configurable") and c["enabled"]: continue state = "开" if c["enabled"] else "关" kind = "硬" if c["kind"] == "hard" else "软" flag = "" if c.get("configurable") else "（内建）" lines.append(f"· [{c['code']}] {c['name']}：{state}/{kind}{flag}") text = f"约束剖面【{profile['name']}】：\n" + "\n".join(lines) text += "\n可说「关闭物料齐套」「齐套改为硬约束」或在设置 → 约束配置调整。" return AgentReply(text=text) if name == "constraint.profile.save": from server.aps_domain.constraints import confirmation_for_profile_save raw = dict(intent.params or {}) payload: dict[str, Any] = {} if raw.get("resetDefaults"): payload["resetDefaults"] = True elif raw.get("constraints"): payload["constraints"] = raw["constraints"] else: cid = str(raw.get("constraintId") or "") if not cid: return AgentReply(text="请指定约束，例如：「关闭物料齐套」或「齐套改为硬约束」。") patch: dict[str, Any] = {} if "enabled" in raw: patch["enabled"] = bool(raw["enabled"]) if raw.get("kind") in ("hard", "soft"): patch["kind"] = raw["kind"] payload["constraints"] = {cid: patch} try: title, lines = confirmation_for_profile_save(store.data, payload) except ValueError as exc: return AgentReply(text=f"约束配置有误：{exc}") block = harness.stage_confirmation( session_id, "constraint.profile.save", payload, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="constraint.profile.save.stage", target={"type": "CONSTRAINT_PROFILE", "id": "default"}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block]) # ---- 约束规则库：默认通用规则 + 计划员自然语言扩充（SC-04） ---- if name == "constraint.rule.query": from server.aps_domain.constraint_rules import constraint_library lib = constraint_library(store.data, getattr(store, "tenant_uuid", "platform")) s = lib["summary"] lines = [ f"· {r['name']}（{r['categoryLabel']}／{r['scopeLabel']}／{r['kindLabel']}）" f"→ {r['effect']['text']}" for r in lib["custom"] ] text = (f"默认通用约束 {s['builtinTotal']} 条（无需逐条登记）；" f"已扩充 {s['customTotal']} 条，其中接入排产引擎 {s['engineBound']} 条、" f"仅记录 {s['recordOnly']} 条。") text += "\n" + ("\n".join(lines) if lines else "还没有扩充规则。直接用一句话说你的要求，我来整理成规则。") return AgentReply(text=text) if name == "constraint.rule.save": from server.aps_domain.constraint_rules import confirmation_for_rule_save raw = dict(intent.params or {}) try: title, lines = confirmation_for_rule_save( store.data, raw, getattr(store, "tenant_uuid", "platform")) except ValueError as exc: return AgentReply(text=f"规则没有登记：{exc}") block = harness.stage_confirmation( session_id, "constraint.rule.save", raw, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="constraint.rule.save.stage", target={"type": "CONSTRAINT_RULE", "id": str(raw.get("ruleId") or "new")}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title} 属于 P2 写操作，需要你确认。", blocks=[block]) # ---- IND-02 SOP→约束 ---- if name == "sop.compile": from server.aps_domain.sop_rules import compile_sop_by_asset, list_compilable_sops, sop_as_block asset_id = str((intent.params or {}).get("assetId") or (intent.params or {}).get("asset_id") or "").strip() if not asset_id: return AgentReply( text="请先选择要查看的作业标准，本次未从问句自动选择。" ) compiled = compile_sop_by_asset(get_knowledge().assets, asset_id) if not compiled.get("ok"): avail = compiled.get("available") or list_compilable_sops(get_knowledge().assets) titles = "、".join(a.get("title") or "?" for a in avail[:6]) return AgentReply(text=f"{compiled.get('error')}\n可编译 SOP：{titles or '（无）'}") pack = compiled["pack"] write_audit(store.data, store.next_id, actor=actor, category="ALGO_RUN", action="sop.compile", target={"type": "SOP", "id": (pack.get("source") or {}).get("assetId")}, power="P0", rationale={"assetId": asset_id, "effects": len(pack.get("effects") or [])}) store.save() text = ( f"已从作业标准《{pack.get('title')}》识别出以下排产规则：\n" f"{pack.get('plannerSummary') or '当前作业标准暂未识别出可用于排产的约束'}\n" "需要应用时请回复「应用该作业标准」，确认后生效。" ) return AgentReply(text=text, blocks=[sop_as_block(pack)]) if name == "sop.apply": from server.aps_domain.sop_rules import ( compile_sop_by_asset, confirmation_for_sop_apply, ) raw = dict(intent.params or {}) asset_id = str(raw.get("assetId") or raw.get("asset_id") or "").strip() pack = raw.get("pack") if isinstance(raw.get("pack"), dict) else None if pack is None: if not asset_id: return AgentReply( text="请先选择要应用的作业标准，本次未生成确认项。" ) compiled = compile_sop_by_asset(get_knowledge().assets, asset_id) if not compiled.get("ok"): return AgentReply(text="作业标准读取失败，请稍后重试。") pack = compiled["pack"] title, lines = confirmation_for_sop_apply(pack) block = harness.stage_confirmation( session_id, "sop.apply", {"pack": pack, "assetId": asset_id or None}, title=title, summary_lines=lines) write_audit(store.data, store.next_id, actor=actor, category="GATE", action="sop.apply.stage", target={"type": "RULE_PACK", "id": pack.get("packId")}, power="P2", rationale={"confirmId": block.props["confirmId"]}) store.save() return AgentReply(text=f"{title}，需要你确认后生效。", blocks=[block]) # ---- 帮助（P0） ---- if name == "help": return AgentReply(text=_HELP) # 未登记工具：Pi 只能调用封闭工具目录；这里显式拒绝，不做自然语言兜底。 return AgentReply(text="这项操作不在当前可执行工具目录中，本次未执行任何操作。") # ============================================================ # 自动化规则动作 → 真实业务动作 执行桥（round-40 方向 S · 矩阵 76） # 只新增段落：commit → schedule.publish（发布）、reschedule → flex.reschedule（重排）。 # 门禁路径与 handle_intent 分支一致（_stage_publish / _flex_reschedule 出 P2 确认卡）； # auto=True（G4 受控自动，AutomationGate 升权凭据已过）时出卡后立即经 # execute_confirmed 批准执行——真实写入始终走既有 harness 门禁。 # ============================================================ def _confirm_id_of(reply: AgentReply) -> str | None: """从确认卡回复块中提取 confirmId；无确认卡返回 None。""" for block in reply.blocks or []: props = getattr(block, "props", None) or {} if getattr(block, "type", None) == "confirm-card" and props.get("confirmId"): return str(props["confirmId"]) return None def run_automation_intent(store: WorldStore, session_id: str, intent: IntentResult, actor: str = "automation", *, auto: bool = False) -> dict[str, Any]: """执行一条自动化规则动作绑定的真实业务意图（矩阵 76 方向 S）。 - auto=False（G2 沙盒 / G3 监督）：与 handle_intent 相同分支出 P2 确认卡（STAGED）； - auto=True（G4 受控自动，AutomationGate 升权凭据已过）：出卡后立即经 execute_confirmed 批准执行（EXECUTED）——真实写入始终过既有 harness 门禁。 未接线的意图抛 ValueError（fail closed，不产生任何写入）。 """ name = intent.intent if name == "schedule.publish": reply = _stage_publish(store, session_id, actor) # handle_intent 同款分支 elif name == "flex.reschedule": reply = _flex_reschedule(store, intent, session_id, actor) # handle_intent 同款分支 else: raise ValueError(f"自动化规则未接线的真实业务意图：{name}") confirm_id = _confirm_id_of(reply) if auto and confirm_id: text = execute_confirmed(store, confirm_id, True, actor=actor) return {"status": "EXECUTED", "text": text, "confirmId": confirm_id} if confirm_id: return {"status": "STAGED", "text": reply.text, "confirmId": confirm_id} return {"status": "PROPOSED", "text": reply.text}