From 47ef7bbed9493b7e27c512052a1793a60538b11c Mon Sep 17 00:00:00 2001 From: chenpei Date: Tue, 8 Sep 2026 01:42:41 +0800 Subject: [PATCH] feat: harden CAD resume and hardware validation --- .../ai-solid-wood-cost/VALIDATION_LEDGER.md | 14 + .../manufacturing_oracle.py | 78 ++ .../probe_structure_graph.py | 744 +++++++++++++++++- .../ai-solid-wood-cost/structure_graph.py | 44 +- .../test_manufacturing_oracle.py | 273 ++++++- .../test_structure_graph.py | 13 +- 6 files changed, 1143 insertions(+), 23 deletions(-) diff --git a/server_patch/ai_3d_model/ai-solid-wood-cost/VALIDATION_LEDGER.md b/server_patch/ai_3d_model/ai-solid-wood-cost/VALIDATION_LEDGER.md index 4996215..a825ee5 100644 --- a/server_patch/ai_3d_model/ai-solid-wood-cost/VALIDATION_LEDGER.md +++ b/server_patch/ai_3d_model/ai-solid-wood-cost/VALIDATION_LEDGER.md @@ -265,3 +265,17 @@ - v104启动前服务器只读检查为`active=0 waiting=0 dewu_busy=0 dewu_queue=0 durable_inflight=0`。从同一已审计v96检查点恢复，只执行一轮：基线3个编译阻塞和1个真实B-rep冲突；AI候选造成11个编译问题被拒绝；随后进入一次聚焦原图裁决，没有继续前后移动循环。 - v104多模态模型连续2次仍因选项过长未通过输出校验，系统明确记录`focused_source_adjudication_error`，并使用`deterministic_dead_end_question_fallback`安全收口为一个问题：“尺寸图标注的总深 400mm，是否包含柜门厚度？”，选项仅为“包含柜门（成品最外尺寸）”与“不包含柜门（只算柜体/框架）”。报告位于服务器隔离目录`/tmp/dalongxia-cad-v85-eI8HeF/probe-sideboard19/sideboard/report.json`。 - 本地新增参考尺寸证据门、问题优先级、焦点不匹配拒绝和确定性兜底专项检查后，结构图谱/制造层74项通过；最终候选完整回归210项通过（235.977秒）。v104仍不等于图像裁决或制造模型通过，生产服务未改、未重启。 + +## 2026-09-08 v105-v112 用户事实续跑、视觉复核净化与五金安装原子修复（候选，未上线） + +- 续跑接口新增结构化 `scope_facts`。用户已经回答“总深不包含柜门”后，该事实绑定到原尺寸证据的 Y 轴，程序把 400mm 柜体参考范围投影为门前沿 0mm、柜体 20–420mm、实际占用深度 420mm；只有端到端阻塞总数严格下降才接受，重复续跑保持幂等，不再重复询问同一个口径问题。 +- v105 暴露旧续跑仍会重新询问已回答问题；修复后，阻塞清空时直接进入制造编译、真实 STEP 构建和视觉复核，不再发起没有待修项的模型“修复”。 +- v106 在尺寸口径解除后收敛到 3 个问题：层板与背板的非结构接触被误生成连接，以及背板先被左右侧板加宽后，顶/底板槽仍使用旧净宽，造成两处真实 B-rep 干涉。后续发散轮次被事务门拒绝，没有覆盖较好检查点。 +- 制造规划改为只在被捕获薄板的边真正进入另一板件宽面时生成槽；层板/隔板边仅抵到背板宽面时记为 `non_structural_abutment`，不产生虚构螺钉、槽或成本。所有入槽余量累计完成后，再按最终嵌入板制造尺寸重算每个承载槽，消除迭代顺序导致的槽宽不足。v107 对 v106 最佳检查点重新编译后，结构、制造和 B-rep 阻塞均为 0。 +- v108 隔离进程未继承视觉模型凭据，旧逻辑把配置缺失误当几何错误并进入修复循环。现在将其分类为不可重试的 `model_credentials_missing`，交给运行时配置层处理，不污染图谱，也不重复请求模型。 +- v109 的视觉模型输出自相矛盾：一边声称层板、铰链、旋钮无需修改，一边把它们列为问题；还试图用透视照片估算覆盖尺寸图明确给出的 40mm 腿高。视觉复核现仅接受枚举问题码、有效原图证据、可执行修正；尺寸差异必须同时给出 `expected_mm` 和 `actual_mm` 且真实不一致。声明“无需修改”的行、超长推理文本和无数值尺寸意见全部丢弃并记录审计，精确尺寸账本优先于照片透视感受。 +- 人工查看 v109 四视图发现右门两只铰链仍位于中缝，尽管编译器把它们声明为安装在右侧板。根因是旧 `actual_embed` 使用硬件到安装面的无界距离，远离宿主 528mm 的铰链也能伪装成贯穿安装。现改为硬件与宿主厚度区间的真实交集并限制最大嵌入深度；只有存在实际空间接触的安装面才兼容。v110 因此稳定阻断两只错误铰链，未被 B-rep“无冲突”假通过。 +- v111 证明只给模型安装方式候选仍不足：模型只改 `through/embed/face` 声明而不移动实体，事务门正确拒绝。系统随后为无简单候选的五金生成原子候选集，把单轴位置、安装方式、安装面和嵌入深度绑成一组；每组必须先经完整图谱重编译并严格减少阻塞，视觉模型只能从原图支持的整组候选中选择，不能仅改声明绕过校验。 +- v112 从同一审计检查点选择右侧板候选后，结构、制造、B-rep和净化后的视觉阻塞均为 0；报告状态为 `candidate_visual_pass_requires_independent_acceptance`，没有自动宣告成功。人工独立查看最终正视和等轴视图后仍判定不合格：铰链虽从中缝移到左右外侧，但被画成横跨门板/侧板的长方体“贯穿件”，与家具暗铰链的铰杯、底座及孔位工艺不符。这证明几何合法不等于五金工艺语义正确。 +- 本轮明确记录的失败路线：让模型只修改五金安装元数据；用通用长方体代替具体铰链家族；把视觉模型无证据的透视估算当尺寸修正；把运行时凭据缺失交给几何重试。后续不得重复这些做法，应增加有证据的五金家族/安装模板（例如铰杯直径与深度、门板杯孔、侧板底座孔距）并用原图或用户确认选择具体模板。 +- v112 仍只是一组人工生成样例的隔离验证，未使用公司 BOM 参与推理、未修改生产服务。不同公司产品盲测、组装/零件外观独立验收、生产级五金孔位与冻结报价后总成本误差不超过 5%仍未完成，接口状态必须保持测试失败/继续优化。 diff --git a/server_patch/ai_3d_model/ai-solid-wood-cost/manufacturing_oracle.py b/server_patch/ai_3d_model/ai-solid-wood-cost/manufacturing_oracle.py index 2c32693..5e2b3f7 100644 --- a/server_patch/ai_3d_model/ai-solid-wood-cost/manufacturing_oracle.py +++ b/server_patch/ai_3d_model/ai-solid-wood-cost/manufacturing_oracle.py @@ -150,6 +150,20 @@ def _method_for(contact: dict, instances: dict, policy: dict) -> str: roles = {a.get("functional_role"), b.get("functional_role")} materials = {a.get("material_family"), b.get("material_family")} if roles & CAPTURED_ROLES: + # A back/drawer-bottom is captured only when *its edge* enters the + # broad face of another panel. When a shelf/divider edge merely + # terminates against the broad face of a back, that contact does not + # imply another groove, screw or machining charge. + if contact.get("interface_type") == "edge_to_face": + edge = instances.get(str(contact.get("edge_instance_id") or "")) + broad = instances.get(str(contact.get("broad_face_instance_id") or "")) + if edge and edge.get("functional_role") in CAPTURED_ROLES: + return str(policy["captured_panel_joint"]) + if ( + broad and broad.get("functional_role") in CAPTURED_ROLES + and edge and edge.get("functional_role") in {"shelf", "divider"} + ): + return "non_structural_abutment" return str(policy["captured_panel_joint"]) if "shelf" in roles: return str(policy["shelf_joint"]) @@ -262,6 +276,10 @@ def plan_manufacturing_features(compiled: dict, *, policy: dict | None = None) - interface_type=contact.get("interface_type"), material_families=[a["material_family"], b["material_family"]]) connection_plans.append(plan) continue + if method == "non_structural_abutment": + plan["feature_source"] = "verified_contact_requires_no_independent_connector" + connection_plans.append(plan) + continue if method == "captured_groove": insert = a if a.get("functional_role") in CAPTURED_ROLES else b if b.get("functional_role") in CAPTURED_ROLES else None receiver_id = contact.get("broad_face_instance_id") @@ -384,6 +402,66 @@ def plan_manufacturing_features(compiled: dict, *, policy: dict | None = None) - ] connection_plans.append(plan) + # Capture allowances accumulate in more than one axis. For example, a + # back captured by the two sides is wider than its net visible width when + # it subsequently enters the top/bottom grooves. Refit every groove to + # the *final* manufacturing insert geometry so the cut fully clears the + # same body exported to STEP. This is a second pass deliberately: the + # result must not depend on derived-contact iteration order. + for receiver_id, instance_features in features.items(): + receiver = instances[receiver_id] + receiver_low = [float(value) for value in receiver["position_mm"]] + receiver_high = [ + receiver_low[axis] + float(receiver["size_mm"][axis]) for axis in range(3) + ] + for groove in instance_features["grooves"]: + if groove.get("validation_status") != "valid": + continue + insert = instances.get(str(groove.get("insert_instance_id") or "")) + adjusted = manufacturing_geometry.get(str(groove.get("insert_instance_id") or "")) + if not insert or not adjusted: + groove["validation_status"] = "blocked" + issue( + "groove_insert_geometry", + "槽口无法取得最终嵌入板制造尺寸", + instance_id=receiver_id, + feature_id=groove["feature_id"], + ) + continue + normal_axis = _axis_index(str(groove["face"])) + thickness_axis = "xyz".index(str(insert["thickness_axis"])) + center = [float(groove["world_center_mm"][name]) for name in "xyz"] + size = [float(groove["world_size_mm"][name]) for name in "xyz"] + insert_low = [float(value) for value in adjusted["position_mm"]] + insert_high = [ + insert_low[axis] + float(adjusted["size_mm"][axis]) for axis in range(3) + ] + clearance = float(groove.get("clearance_mm") or 0.0) + valid = True + for axis in range(3): + if axis == normal_axis: + continue + extra = clearance / 2.0 if axis == thickness_axis else 0.0 + low = max(receiver_low[axis], insert_low[axis] - extra) + high = min(receiver_high[axis], insert_high[axis] + extra) + if high - low <= 0.01: + valid = False + break + center[axis] = (low + high) / 2.0 + size[axis] = high - low + if not valid: + groove["validation_status"] = "blocked" + issue( + "groove_final_coverage", + "最终嵌入板与承载件没有可加工的槽口交集", + instance_id=receiver_id, + feature_id=groove["feature_id"], + ) + continue + groove["world_center_mm"] = _world_dict(center) + groove["world_size_mm"] = _world_dict(size) + groove["coverage_source"] = "final_insert_geometry_after_all_capture_allowances" + # A duplicated operation would be machined and charged twice. Detect it # by geometry and instance, independent of feature names. for instance_id, rows in features.items(): diff --git a/server_patch/ai_3d_model/ai-solid-wood-cost/probe_structure_graph.py b/server_patch/ai_3d_model/ai-solid-wood-cost/probe_structure_graph.py index 3807d8b..b6a2d66 100644 --- a/server_patch/ai_3d_model/ai-solid-wood-cost/probe_structure_graph.py +++ b/server_patch/ai_3d_model/ai-solid-wood-cost/probe_structure_graph.py @@ -10,6 +10,7 @@ import base64 import copy import importlib.util import json +import math import os import pathlib import re @@ -37,6 +38,210 @@ class AuthoritativeKernelError(RuntimeError): self.feedback = feedback +def terminal_dependency_feedback(error: Exception | str) -> dict | None: + """Classify non-geometric configuration failures that retries cannot fix.""" + message = str(error or "").strip() + if "尚未配置密钥" in message: + return { + "code": "model_credentials_missing", + "severity": "blocking", + "detail": "隔离验证进程未取得读图模型凭据", + "next_owner": "runtime_configuration", + "retryable": False, + } + return None + + +VISUAL_ISSUE_CODES = { + "missing_component", "extra_component", "component_count_mismatch", + "geometric_placement_mismatch", "orientation_mismatch", + "assembly_relation_mismatch", "silhouette_mismatch", + "numeric_dimension_mismatch", "hardware_type_mismatch", + "hardware_count_mismatch", "door_type_mismatch", +} + + +def normalize_visual_review( + value: dict, *, image_ids: set[str], dimension_ledger: dict, +) -> dict: + """Reject self-contradictory or non-actionable visual-review prose.""" + if not isinstance(value, dict): + raise ValueError("视觉复核必须返回JSON对象") + audit = [] + issues = [] + raw_issues = value.get("issues") or [] + if not isinstance(raw_issues, list) or len(raw_issues) > 16: + raise ValueError("视觉复核问题列表无效") + for index, row in enumerate(raw_issues): + if not isinstance(row, dict): + audit.append({"index": index, "accepted": False, "reason": "issue_not_object"}) + continue + code = str(row.get("code") or "").strip().lower() + if code == "dimension_mismatch": + code = "numeric_dimension_mismatch" + component = str(row.get("component") or "").strip() + image_id = str(row.get("source_image_id") or "").strip() + evidence = str(row.get("evidence") or "").strip() + correction = str(row.get("correction") or "").strip() + combined = f"{evidence}\n{correction}".lower() + if code not in VISUAL_ISSUE_CODES: + audit.append({"index": index, "accepted": False, "reason": "unsupported_issue_code", "code": code}) + continue + if not component or image_id not in image_ids or not evidence or not correction: + audit.append({"index": index, "accepted": False, "reason": "missing_actionable_evidence", "code": code}) + continue + if any(phrase in combined for phrase in ( + "no correction needed", "no major issue", "无需修改", "不需要修改", "不需修正", + )): + audit.append({"index": index, "accepted": False, "reason": "self_declared_non_issue", "code": code}) + continue + if len(evidence) > 600 or len(correction) > 400: + audit.append({"index": index, "accepted": False, "reason": "overlong_reasoning_not_diagnostic", "code": code}) + continue + cleaned = { + "code": code, "component": component, "source_image_id": image_id, + "evidence": evidence, "correction": correction, + } + if code == "numeric_dimension_mismatch": + try: + expected = float(row.get("expected_mm")) + actual = float(row.get("actual_mm")) + except (TypeError, ValueError): + audit.append({"index": index, "accepted": False, "reason": "numeric_values_required", "code": code}) + continue + if not math.isfinite(expected) or not math.isfinite(actual) or abs(expected - actual) <= 0.5: + audit.append({"index": index, "accepted": False, "reason": "numeric_mismatch_not_proven", "code": code}) + continue + cleaned.update({"expected_mm": expected, "actual_mm": actual}) + issues.append(cleaned) + audit.append({"index": index, "accepted": True, "code": code}) + uncertain = [] + raw_uncertain = value.get("uncertain") or [] + if not isinstance(raw_uncertain, list) or len(raw_uncertain) > 8: + raise ValueError("视觉复核不确定项列表无效") + for index, row in enumerate(raw_uncertain): + if not isinstance(row, dict): + continue + image_id = str(row.get("source_image_id") or "").strip() + question = str(row.get("single_user_question") or "").strip() + evidence = str(row.get("evidence") or "").strip() + if image_id in image_ids and question and evidence and len(question) <= 80 and len(evidence) <= 400: + uncertain.append({ + "code": str(row.get("code") or "visual_uncertainty").strip(), + "component": str(row.get("component") or "").strip(), + "source_image_id": image_id, + "evidence": evidence, + "single_user_question": question, + }) + return { + "ok": not issues and not uncertain, + "issues": issues, + "uncertain": uncertain, + "normalization_audit": audit, + "dimension_ledger_axes": sorted(axis for axis in "xyz" if axis in (dimension_ledger or {})), + } + + +def enrich_hardware_feedback_with_verified_placements( + graph: dict, compiled: dict, feedback: list[dict], *, compile_candidate, +) -> list[dict]: + """Attach atomic, compiler-proven placement candidates to mounting errors. + + The candidates are not applied automatically because multiple physically + valid mounting faces can exist. The image model must select the candidate + supported by the source views, but it cannot repair only the declaration + while leaving the hardware body hundreds of millimetres away. + """ + result = copy.deepcopy(feedback) + solved = {str(row.get("id")): row for row in compiled.get("instances") or []} + current_issue_count = len(compiled.get("issues") or []) + envelope = list(compiled.get("occupied_envelope_mm") or [0, 0, 0]) + for item in result: + if ( + item.get("code") != "hardware_mounting_geometry" + or item.get("compatible_mounting_values") + ): + continue + instance_id = str(item.get("instance_id") or "") + host_id = str(item.get("host_instance_id") or "") + hardware, host = solved.get(instance_id), solved.get(host_id) + path = str(item.get("path") or "") + match = re.search(r"/mountings/(\d+)$", path) + axis_name = str((host or {}).get("thickness_axis") or "") + if not hardware or not host or not match or axis_name not in "xyz": + continue + mounting_index = int(match.group(1)) + axis = "xyz".index(axis_name) + hardware_size = float(hardware["size_mm"][axis]) + host_low = float(host["position_mm"][axis]) + host_thickness = float(host["size_mm"][axis]) + host_high = host_low + host_thickness + current_position = float(hardware["position_mm"][axis]) + candidate_specs = [] + for face in (f"{axis_name}-", f"{axis_name}+"): + surface_position = host_low - hardware_size if face.endswith("-") else host_high + candidate_specs.append((face, "surface", 0.0, surface_position)) + if hardware_size + 0.01 >= host_thickness: + through_position = host_high - hardware_size if face.endswith("-") else host_low + candidate_specs.append((face, "through", host_thickness, through_position)) + if item.get("kind") == "mortised": + depth = float(item.get("declared_embed_depth_mm") or 0.0) + if 0.6 < depth < min(host_thickness, hardware_size) - 0.6: + candidate_specs.extend([ + (f"{axis_name}-", "mortised", depth, host_low - (hardware_size - depth)), + (f"{axis_name}+", "mortised", depth, host_high - depth), + ]) + candidates = [] + seen = set() + for face, kind, depth, position in candidate_specs: + signature = (face, kind, round(depth, 6), round(position, 6)) + if signature in seen: + continue + seen.add(signature) + patches = [ + {"op": "replace", "path": f"/instances/{instance_id}/placement/{axis_name}", + "value": {"mode": "absolute", "value_mm": round(position, 6)}}, + {"op": "replace", "path": f"/instances/{instance_id}/mountings/{mounting_index}/kind", "value": kind}, + {"op": "replace", "path": f"/instances/{instance_id}/mountings/{mounting_index}/face", "value": face}, + {"op": "replace", "path": f"/instances/{instance_id}/mountings/{mounting_index}/embed_depth_mm", "value": round(depth, 6)}, + ] + try: + candidate_graph, _ = apply_graph_patch(graph, patches) + candidate_compiled = compile_candidate(candidate_graph) + except Exception: + continue + candidate_issues = list(candidate_compiled.get("issues") or []) + if len(candidate_issues) >= current_issue_count: + continue + if any( + row.get("code") == "hardware_mounting_geometry" and row.get("path") == path + for row in candidate_issues + ): + continue + outside = 0.0 + if len(envelope) == 3: + outside = max(0.0, -position) + max(0.0, position + hardware_size - float(envelope[axis])) + candidates.append({ + "patches": patches, + "candidate_compile_issue_count": len(candidate_issues), + "candidate_issue_codes": sorted({str(row.get("code")) for row in candidate_issues}), + "outside_occupied_envelope_mm": round(outside, 6), + "movement_mm": round(abs(position - current_position), 6), + "validation": "full_graph_recompile_strict_issue_reduction", + }) + candidates.sort(key=lambda row: ( + row["candidate_compile_issue_count"], row["outside_occupied_envelope_mm"], + row["movement_mm"], json.dumps(row["patches"], sort_keys=True), + )) + if candidates: + item["compatible_placement_patches"] = candidates[:4] + item["correction"] = ( + "compatible_mounting_values为空，必须从compatible_placement_patches中回看原图选择一组" + "完整原子补丁；不得只修改kind/face/embed_depth_mm而不移动五金实体。" + ) + return result + + def source_for_role(asset_dir: pathlib.Path, case: str, role: str) -> pathlib.Path | None: """Resolve test evidence by semantic role, not one fragile filename.""" aliases = ROLE_FILENAME_ALIASES[role] @@ -346,6 +551,230 @@ def deterministic_dead_end_question( } +def normalize_scope_facts(value, *, image_ids: set[str], dimension_ledger: dict) -> list[dict]: + """Validate externally supplied dimension-scope facts for an isolated resume. + + These facts represent either an exact user answer already captured by the + conversation runtime or an independent source adjudication. They are not + model-authored graph content and therefore remain outside the patch loop. + """ + rows = value if isinstance(value, list) else [value] + if not rows or len(rows) > 3: + raise ValueError("尺寸口径事实必须包含1至3条记录") + cleaned = [] + seen_axes = set() + for row in rows: + if not isinstance(row, dict): + raise ValueError("每条尺寸口径事实必须是JSON对象") + fact_id = str(row.get("id") or "").strip() + source_type = str(row.get("source_type") or "").strip() + axis = str(row.get("axis") or "").strip() + ledger_scope = str(row.get("ledger_scope") or "").strip() + confidence = row.get("confidence") + source_ids = row.get("source_image_ids") or [] + text = str(row.get("text") or "").strip() + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:-]{2,79}", fact_id): + raise ValueError("尺寸口径事实缺少稳定ID") + if source_type not in {"user", "independent_source_adjudication"}: + raise ValueError("尺寸口径事实来源必须是用户确认或独立原图裁决") + if axis not in {"x", "y", "z"} or axis in seen_axes: + raise ValueError("每个轴只允许一条尺寸口径事实") + if ledger_scope not in {"occupied_overall", "reference"}: + raise ValueError("尺寸口径只能是实际占用外形或柜体/开口参考尺寸") + if axis not in (dimension_ledger or {}): + raise ValueError("尺寸口径事实必须对应已通过资格门的原图尺寸轴") + if ( + isinstance(confidence, bool) or not isinstance(confidence, (int, float)) + or not 0.8 <= float(confidence) <= 1.0 + ): + raise ValueError("尺寸口径事实置信度必须不低于0.8") + if source_type == "user" and float(confidence) != 1.0: + raise ValueError("用户明确确认的尺寸口径必须按1.0记录") + if ( + not isinstance(source_ids, list) or not source_ids + or not set(source_ids).issubset(image_ids) + ): + raise ValueError("尺寸口径事实必须引用本轮真实原图ID") + if not text or len(text) > 300: + raise ValueError("尺寸口径事实必须保存简短原始依据") + cleaned.append({ + "id": fact_id, "source_type": source_type, "axis": axis, + "ledger_scope": ledger_scope, "confidence": round(float(confidence), 4), + "source_image_ids": sorted(set(source_ids)), "text": text, + }) + seen_axes.add(axis) + return cleaned + + +def bind_scope_facts(graph: dict, scope_facts: list[dict]) -> tuple[dict, dict, dict, list[dict]]: + """Bind trusted facts without allowing the graph model to rewrite them.""" + result = copy.deepcopy(graph) + observations = list(result.get("observations") or []) + user_evidence = {} + adjudications = {} + audit = [] + fact_ids = {row["id"] for row in scope_facts} + observations = [row for row in observations if str(row.get("id") or "") not in fact_ids] + overrides = copy.deepcopy(result.get("dimension_scope_overrides") or {}) + if not isinstance(overrides, dict): + overrides = {} + for fact in scope_facts: + fact_id, axis, scope = fact["id"], fact["axis"], fact["ledger_scope"] + if fact["source_type"] == "user": + observation = { + "id": fact_id, "source_type": "user", "text": fact["text"], + "confidence": 1.0, + } + observations.append(observation) + user_evidence[fact_id] = copy.deepcopy(observation) + reference = {"ledger_scope": scope, "evidence_id": fact_id} + else: + adjudications[fact_id] = { + "axis": axis, "ledger_scope": scope, + "confidence": fact["confidence"], + "source_image_ids": list(fact["source_image_ids"]), + "evidence": fact["text"], + } + reference = {"ledger_scope": scope, "adjudication_id": fact_id} + if scope == "reference": + overrides[axis] = reference + else: + overrides.pop(axis, None) + audit.append({ + "fact_id": fact_id, "axis": axis, "ledger_scope": scope, + "source_type": fact["source_type"], + "binding": "authoritative_outside_model_patch_loop", + }) + result["observations"] = observations + if overrides: + result["dimension_scope_overrides"] = overrides + else: + result.pop("dimension_scope_overrides", None) + return result, user_evidence, adjudications, audit + + +def propose_reference_scope_projection( + graph: dict, compiled: dict, scope_facts: list[dict], dimension_ledger: dict, +) -> tuple[dict | None, list[dict]]: + """Propose a bounded overlay-front coordinate-frame expansion. + + This is a candidate, never an unconditional repair. It is generated only + when a verified fact says an overall label excludes a door/drawer front, + the front instances all touch the same current envelope face, and their + thickness axis matches that label. The normal end-to-end score and B-rep + checks must still improve before the caller accepts it. + """ + result = copy.deepcopy(graph) + envelope = list(compiled.get("occupied_envelope_mm") or []) + compiled_instances = list(compiled.get("instances") or []) + graph_instances = {str(row.get("id")): row for row in result.get("instances") or []} + graph_assemblies = {str(row.get("id")): row for row in result.get("assemblies") or []} + if len(envelope) != 3: + return None, [{"accepted": False, "reason": "occupied_envelope_unavailable"}] + reference_bounds = copy.deepcopy(result.get("reference_bounds_mm") or { + axis: [0, envelope[index]] for index, axis in enumerate("xyz") + }) + audit = [] + changed = False + for fact in scope_facts: + if fact.get("ledger_scope") != "reference": + continue + axis = fact["axis"] + axis_index = "xyz".index(axis) + ledger_fact = (dimension_ledger or {}).get(axis) or {} + reference_size = ledger_fact.get("value_mm") + if isinstance(reference_size, bool) or not isinstance(reference_size, (int, float)): + audit.append({"fact_id": fact["id"], "accepted": False, "reason": "qualified_size_missing"}) + continue + fronts = [ + row for row in compiled_instances + if row.get("functional_role") in {"door", "drawer_front"} + and row.get("thickness_axis") == axis + ] + if not fronts: + audit.append({"fact_id": fact["id"], "accepted": False, "reason": "matching_front_instance_missing"}) + continue + min_touch = all(abs(float(row["position_mm"][axis_index])) <= 0.5 for row in fronts) + max_touch = all( + abs(float(row["position_mm"][axis_index]) + float(row["size_mm"][axis_index]) - float(envelope[axis_index])) <= 0.5 + for row in fronts + ) + if min_touch == max_touch: + audit.append({"fact_id": fact["id"], "accepted": False, "reason": "front_envelope_face_ambiguous"}) + continue + projection = max(float(row["size_mm"][axis_index]) for row in fronts) + if projection <= 0 or projection > 100: + audit.append({"fact_id": fact["id"], "accepted": False, "reason": "front_projection_out_of_range"}) + continue + actual_size = float(reference_size) + projection + expected_reference_bounds = [projection if min_touch else 0.0, actual_size if min_touch else float(reference_size)] + existing_reference_bounds = reference_bounds.get(axis) + if ( + abs(float(envelope[axis_index]) - actual_size) <= 0.01 + and isinstance(existing_reference_bounds, list) + and len(existing_reference_bounds) == 2 + and all( + abs(float(existing_reference_bounds[index]) - expected_reference_bounds[index]) <= 0.01 + for index in range(2) + ) + ): + audit.append({ + "fact_id": fact["id"], "accepted": False, "axis": axis, + "reason": "reference_scope_projection_already_bound", + "occupied_size_mm": round(actual_size, 4), + "reference_bounds_mm": [round(value, 4) for value in expected_reference_bounds], + }) + continue + new_envelope = list(envelope) + new_envelope[axis_index] = actual_size + result["envelope_mm"] = [round(value, 4) for value in new_envelope] + reference_bounds = { + name: [0, round(new_envelope[index], 4)] for index, name in enumerate("xyz") + } + shift = projection if min_touch else 0.0 + reference_bounds[axis] = [round(shift, 4), round(shift + float(reference_size), 4)] + + front_assembly_ids = {str(row.get("assembly_id") or "") for row in fronts} + shifted_assembly_ids = set() + if min_touch: + for assembly_id, assembly in graph_assemblies.items(): + if assembly_id in front_assembly_ids or assembly.get("role") in {"door", "hardware_group"}: + continue + bounds = assembly.get("bounds_mm") or {} + values = bounds.get(axis) + if not isinstance(values, list) or len(values) != 2: + continue + try: + low, high = (float(values[0]), float(values[1])) + except (TypeError, ValueError): + audit.append({"fact_id": fact["id"], "accepted": False, "reason": "symbolic_assembly_bounds_require_ai"}) + return None, audit + bounds[axis] = [round(low + shift, 4), round(high + shift, 4)] + shifted_assembly_ids.add(assembly_id) + for instance_id, instance in graph_instances.items(): + assembly_id = str(instance.get("assembly_id") or "") + if assembly_id not in shifted_assembly_ids: + continue + placement = instance.get("placement") or {} + axis_rule = placement.get(axis) + if isinstance(axis_rule, dict) and axis_rule.get("target") == "envelope": + axis_rule["target"] = assembly_id + result["reference_bounds_mm"] = reference_bounds + audit.append({ + "fact_id": fact["id"], "accepted": True, "axis": axis, + "front_face": "min" if min_touch else "max", + "front_projection_mm": round(projection, 4), + "reference_size_mm": round(float(reference_size), 4), + "occupied_size_mm": round(actual_size, 4), + "front_instance_ids": sorted(str(row.get("id")) for row in fronts), + "shifted_assembly_ids": sorted(shifted_assembly_ids), + "status": "candidate_requires_full_compile_kernel_visual_validation", + }) + envelope = new_envelope + changed = True + return (result if changed else None), audit + + def select_resume_checkpoint(rows: list[dict]) -> dict: """Select only an accepted, internally consistent audited checkpoint.""" candidates = [ @@ -605,6 +1034,66 @@ def refresh_compiled_kernel_feedback(ai3d, build_dir: pathlib.Path, compiled: di } +def build_render_and_review( + *, engine, model: str, images: list[dict], ai3d, build_dir: pathlib.Path, + compiled: dict, manufacturing_plan: dict, image_ids: set[str], dimension_ledger: dict, +) -> dict: + """Build one blocker-free graph and compare four CAD views to originals.""" + build_dir.mkdir(parents=True, exist_ok=True) + source = build_dir / "assembly.step.py" + source.write_text(cad_source(compiled, manufacturing_plan), encoding="utf-8") + ai3d._validate_source(source.read_text(encoding="utf-8")) + built = build_and_authoritatively_validate( + ai3d, build_dir, source.name, requested_exports=["glb"] + ) + built["multiview_previews"] = render_multiview(ai3d, build_dir, built["step_path"]) + cad_views = [] + for item in built["multiview_previews"]: + view_path = pathlib.Path(item["path"]) + cad_views.extend([ + {"type": "text", "text": f"候选CAD {item['camera']}正交视图（不是原图证据）："}, + {"type": "image_url", "image_url": { + "url": "data:image/png;base64," + base64.b64encode(view_path.read_bytes()).decode() + }}, + ]) + review = engine._request_model_json({ + "model": model, + "messages": [{"role": "user", "content": images + cad_views + [{ + "type": "text", "text": ( + "必须分别使用CAD正面、右侧、俯视和等轴视图，对照全部原始图核验CAD侧板分段、抽屉排列、" + "所有可见板件数量、门型、部件方向、连接关系和外形比例。" + "这是诊断草图，不能因为内核已通过或画面简单就判正确。" + "先由AI给出可执行correction；uncertain仅保留所有原图和四个CAD视图都不能消解的隐蔽信息。" + "精确尺寸图和已通过资格门的尺寸台账优先于照片透视比例；不能仅因照片看起来更长或更短就否定图纸尺寸。" + "只有发现真实错误时才列入issues，已经正确或无需修改的项目严禁列入。" + "code只能是missing_component、extra_component、component_count_mismatch、geometric_placement_mismatch、" + "orientation_mismatch、assembly_relation_mismatch、silhouette_mismatch、numeric_dimension_mismatch、" + "hardware_type_mismatch、hardware_count_mismatch、door_type_mismatch。" + "numeric_dimension_mismatch必须同时返回expected_mm和actual_mm。" + "evidence不超过600字符，correction不超过400字符；禁止输出思考过程。" + "只返回JSON {ok:boolean,issues:[{code,component,source_image_id,evidence,correction,expected_mm?,actual_mm?}]," + "uncertain:[{code,component,source_image_id,evidence,single_user_question}]}。" + "每个问题引用一张能证明问题的原图ID；没有原图证据的推测放uncertain。\n当前编译器问题：" + + json.dumps(compiled.get("issues") or [], ensure_ascii=False) + + "\n已通过资格门的尺寸台账：" + json.dumps(dimension_ledger, ensure_ascii=False) + + "\n当前由物理实例生成的BOM：" + json.dumps(compiled.get("bom") or [], ensure_ascii=False) + + "\n由真实接触面生成的孔槽与连接件计划：" + json.dumps(manufacturing_plan, ensure_ascii=False) + ), + }]}], + "response_format": {"type": "json_object"}, "temperature": 0.0, + "enable_thinking": True, "thinking_budget": 2048, + "vl_high_resolution_images": True, "max_tokens": 6000, + }, "原图与结构图谱CAD复核失败", max_attempts=2, timeout_seconds=150, + decoder=lambda text: normalize_visual_review( + json.loads(text), image_ids=image_ids, dimension_ledger=dimension_ledger, + )) + return { + "built": built, + "visual_review": review, + "diagnostic_mounting_clearance_audit": diagnostic_mounting_clearance_cuts(compiled)["audit"], + } + + def main(): parser = argparse.ArgumentParser() parser.add_argument("--engine", required=True) @@ -622,6 +1111,13 @@ def main(): choices=("engineered_panel", "solid_wood", "metal", "plastic", "glass", "other"), help="Authoritative user-selected material family; never inferred from the image.", ) + parser.add_argument( + "--scope-facts-json", + help=( + "JSON object/list of already verified user or independent dimension-scope facts. " + "The probe validates and binds them outside the model patch loop." + ), + ) args = parser.parse_args() out = pathlib.Path(args.output) out.mkdir(parents=True, exist_ok=True) @@ -710,6 +1206,27 @@ def main(): source_dimensions = source_qualification.get("dimensions") or {} dimension_ledger = source_qualification.get("dimension_ledger") or {} topology_contract = source_qualification.get("topology_contract") or {} + scope_facts = [] + if args.scope_facts_json: + scope_facts = normalize_scope_facts( + json.loads(args.scope_facts_json), image_ids=image_ids, + dimension_ledger=dimension_ledger, + ) + report["verified_scope_facts"] = copy.deepcopy(scope_facts) + bound_user_evidence = {} + bound_scope_adjudications = {} + + def compile_candidate_graph(candidate_graph: dict) -> dict: + rebound, candidate_user_evidence, candidate_adjudications, _ = bind_scope_facts( + candidate_graph, scope_facts, + ) + return compile_graph( + rebound, source_image_ids=image_ids, + user_evidence=candidate_user_evidence, + dimension_ledger=dimension_ledger, + dimension_scope_adjudications=candidate_adjudications, + topology_contract=topology_contract, + ) def checkpoint(current_row=None): snapshot = json.loads(json.dumps(report, ensure_ascii=False)) @@ -746,6 +1263,7 @@ def main(): "dimension_read": source_dimensions, "dimension_ledger": dimension_ledger, "topology_contract": topology_contract, + "verified_scope_facts": scope_facts, }, ensure_ascii=False) ) graph, feedback = {}, [] @@ -754,9 +1272,14 @@ def main(): round_number_base = 0 if resume_payload: best = select_resume_checkpoint(resume_payload.get("rounds") or []) - graph = copy.deepcopy(best["graph"]) + graph, bound_user_evidence, bound_scope_adjudications, scope_binding_audit = bind_scope_facts( + copy.deepcopy(best["graph"]), scope_facts, + ) + report["scope_binding_audit"] = scope_binding_audit refreshed_compiled = compile_graph( - graph, source_image_ids=image_ids, dimension_ledger=dimension_ledger, + graph, source_image_ids=image_ids, user_evidence=bound_user_evidence, + dimension_ledger=dimension_ledger, + dimension_scope_adjudications=bound_scope_adjudications, topology_contract=topology_contract, ) accepted_compile_issue_count = len(refreshed_compiled.get("issues") or []) @@ -786,6 +1309,10 @@ def main(): + list((best.get("visual_review") or {}).get("issues") or []) + list((best.get("visual_review") or {}).get("uncertain") or []) ) + feedback = enrich_hardware_feedback_with_verified_placements( + graph, refreshed_compiled, feedback, + compile_candidate=compile_candidate_graph, + ) round_number_base = int(best.get("round") or 0) report["resumed_from"] = { "source_round": round_number_base, @@ -799,6 +1326,139 @@ def main(): refreshed_kernel.get("diagnostic_mounting_clearance_audit") or [] ), } + if scope_facts: + projected_graph, projection_audit = propose_reference_scope_projection( + graph, refreshed_compiled, scope_facts, dimension_ledger, + ) + projection_record = {"audit": projection_audit, "accepted": False} + if projected_graph is not None: + projected_graph, projected_user_evidence, projected_adjudications, projected_binding = bind_scope_facts( + projected_graph, scope_facts, + ) + projected_compiled = compile_graph( + projected_graph, source_image_ids=image_ids, + user_evidence=projected_user_evidence, + dimension_ledger=dimension_ledger, + dimension_scope_adjudications=projected_adjudications, + topology_contract=topology_contract, + ) + projected_manufacturing = plan_manufacturing_features( + projected_compiled, + policy={"default_material_family": args.material_family}, + ) + projected_kernel = refresh_compiled_kernel_feedback( + ai3d, case_dir / "scope-projection-kernel", projected_compiled, + projected_manufacturing, + ) + projected_checkpoint = { + "compiled": projected_compiled, + "manufacturing_plan": projected_manufacturing, + "cad_kernel_feedback": list(projected_kernel.get("feedback") or []), + } + projected_score = round_blocker_count(projected_checkpoint) + projection_record.update({ + "binding_audit": projected_binding, + "compiled": projected_compiled, + "manufacturing_plan": projected_manufacturing, + "cad_kernel_feedback": projected_checkpoint["cad_kernel_feedback"], + "end_to_end_blocker_count": projected_score, + "previous_end_to_end_blocker_count": accepted_total_blocker_count, + }) + if ( + accepted_total_blocker_count is None + or projected_score < accepted_total_blocker_count + ): + graph = projected_graph + bound_user_evidence = projected_user_evidence + bound_scope_adjudications = projected_adjudications + refreshed_compiled = projected_compiled + refreshed_manufacturing_plan = projected_manufacturing + accepted_compile_issue_count = len(projected_compiled.get("issues") or []) + accepted_total_blocker_count = projected_score + feedback = ( + list(projected_compiled.get("issues") or []) + + ( + list(projected_manufacturing.get("issues") or []) + if projected_compiled.get("ok") else [] + ) + + projected_checkpoint["cad_kernel_feedback"] + ) + feedback = enrich_hardware_feedback_with_verified_placements( + graph, projected_compiled, feedback, + compile_candidate=compile_candidate_graph, + ) + projection_record["accepted"] = True + projection_record["acceptance_rule"] = "strict_end_to_end_blocker_reduction" + else: + projection_record["rejection_reason"] = "end_to_end_blocker_count_not_reduced" + report["scope_projection"] = projection_record + if ( + accepted_total_blocker_count == 0 + and refreshed_compiled.get("ok") + and refreshed_manufacturing_plan.get("ok") + ): + print(json.dumps({ + "case": case, "phase": "resume_blocker_free_visual_review", + "source_round": round_number_base, + }, ensure_ascii=False), flush=True) + try: + resumed_review = build_render_and_review( + engine=engine, model=args.model, images=images, ai3d=ai3d, + build_dir=case_dir / "resume-visual-review", + compiled=refreshed_compiled, + manufacturing_plan=refreshed_manufacturing_plan, + image_ids=image_ids, dimension_ledger=dimension_ledger, + ) + except Exception as exc: + dependency_feedback = terminal_dependency_feedback(exc) + if dependency_feedback: + report["resume_visual_review"] = { + "dependency_error": dependency_feedback, + } + report["status"] = "visual_review_dependency_failed" + reports.append(report) + (case_dir / "report.json").write_text( + json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" + ) + (out / "report.json").write_text( + json.dumps(reports, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(json.dumps({ + "case": case, "status": report["status"], + "blocking": dependency_feedback, + }, ensure_ascii=False), flush=True) + continue + kernel_feedback = ( + exc.feedback if isinstance(exc, AuthoritativeKernelError) + else structured_kernel_feedback(exc) + ) + report["resume_visual_review"] = { + "error": f"{type(exc).__name__}: {exc}", + "cad_kernel_feedback": kernel_feedback, + } + feedback = kernel_feedback + accepted_total_blocker_count = len(kernel_feedback) + report["status"] = "resume_visual_review_build_failed" + else: + review = resumed_review["visual_review"] + report["resume_visual_review"] = resumed_review + feedback = list(review.get("issues") or []) + list(review.get("uncertain") or []) + accepted_total_blocker_count = len(feedback) + if review.get("ok") is True and not feedback and not graph.get("uncertainties"): + report["status"] = "candidate_visual_pass_requires_independent_acceptance" + reports.append(report) + (case_dir / "report.json").write_text( + json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" + ) + (out / "report.json").write_text( + json.dumps(reports, ensure_ascii=False, indent=2), encoding="utf-8" + ) + print(json.dumps({ + "case": case, "status": report["status"], + "source_round": round_number_base, + }, ensure_ascii=False), flush=True) + continue + report["status"] = "resume_visual_not_passed" for turn in range(max(1, min(args.rounds, 5))): started = time.time() row = {"round": round_number_base + turn + 1} @@ -838,7 +1498,7 @@ def main(): "必须replace整个轴对象，或对align轴用add新增offset_mm。" "合法示例：{\"patches\":[{\"op\":\"replace\",\"path\":\"/instances/door_left_inst/placement/x\"," "\"value\":{\"mode\":\"absolute\",\"value_mm\":20}}]}。" - "路径可修改observations、parameters、envelope_mm、assemblies、parts、instances、connections、assembly_connections、uncertainties、placement_solver、orientation_solver、contact_derivation；" + "路径可修改observations、parameters、envelope_mm、reference_bounds_mm、dimension_scope_overrides、assemblies、parts、instances、connections、assembly_connections、uncertainties、placement_solver、orientation_solver、contact_derivation；" "数组追加使用/-。修改后仍需符合furniture-assembly-graph/1。\n上一版图谱：" + json.dumps(graph, ensure_ascii=False) + evidence_context @@ -861,9 +1521,16 @@ def main(): row["patch_normalization_audit"] = normalization_audit row["normalized_patches"] = normalized_patches candidate_graph, patch_audit = apply_graph_patch(graph, normalized_patches) + candidate_graph, candidate_user_evidence, candidate_scope_adjudications, binding_audit = bind_scope_facts( + candidate_graph, scope_facts, + ) + row["scope_binding_audit"] = binding_audit row["patch_audit"] = patch_audit candidate_compiled = compile_graph( - candidate_graph, source_image_ids=image_ids, dimension_ledger=dimension_ledger, + candidate_graph, source_image_ids=image_ids, + user_evidence=candidate_user_evidence, + dimension_ledger=dimension_ledger, + dimension_scope_adjudications=candidate_scope_adjudications, topology_contract=topology_contract, ) candidate_issue_count = len(candidate_compiled.get("issues") or []) @@ -892,6 +1559,8 @@ def main(): checkpoint(row) else: graph = candidate_graph + bound_user_evidence = candidate_user_evidence + bound_scope_adjudications = candidate_scope_adjudications except GraphError as exc: row["patch_error"] = f"{type(exc).__name__}: {exc}" raw_patches = response.get("patches") if isinstance(response, dict) else response @@ -901,16 +1570,26 @@ def main(): row["checkpoint_phase"] = "invalid_patch_rejected_graph_preserved" checkpoint(row) else: - graph = response + graph, bound_user_evidence, bound_scope_adjudications, binding_audit = bind_scope_facts( + response, scope_facts, + ) + row["scope_binding_audit"] = binding_audit if not row.get("patch_error"): row["graph"] = graph compiled = compile_graph( - graph, source_image_ids=image_ids, dimension_ledger=dimension_ledger, + graph, source_image_ids=image_ids, + user_evidence=bound_user_evidence, + dimension_ledger=dimension_ledger, + dimension_scope_adjudications=bound_scope_adjudications, topology_contract=topology_contract, ) row["compiled"] = compiled accepted_compile_issue_count = len(compiled.get("issues") or []) feedback = list(compiled["issues"]) + feedback = enrich_hardware_feedback_with_verified_placements( + graph, compiled, feedback, + compile_candidate=compile_candidate_graph, + ) manufacturing_plan = plan_manufacturing_features( compiled, policy={"default_material_family": args.material_family}, @@ -964,15 +1643,25 @@ def main(): "所有可见板件数量、门型、部件方向、连接关系和外形比例。" "这是诊断草图，编译错误中的零件可能未渲染；不能因为画面简单就判正确。" "先由AI给出可执行correction；uncertain仅保留所有原图和四个CAD视图都不能消解的隐蔽信息。" - "只返回JSON {ok:boolean,issues:[{code,component,source_image_id,evidence,correction}],uncertain:[]}。" + "精确尺寸图和尺寸台账优先于照片透视比例；已经正确或无需修改的项目严禁列入issues。" + "code只能是missing_component、extra_component、component_count_mismatch、geometric_placement_mismatch、" + "orientation_mismatch、assembly_relation_mismatch、silhouette_mismatch、numeric_dimension_mismatch、" + "hardware_type_mismatch、hardware_count_mismatch、door_type_mismatch。" + "numeric_dimension_mismatch必须同时返回expected_mm和actual_mm；禁止输出思考过程。" + "只返回JSON {ok:boolean,issues:[{code,component,source_image_id,evidence,correction,expected_mm?,actual_mm?}],uncertain:[]}。" "每个问题引用一张能证明问题的原图ID；没有原图证据的推测放uncertain。\n编译器问题：" + json.dumps(compiled["issues"], ensure_ascii=False) + + "\n已通过资格门的尺寸台账：" + json.dumps(dimension_ledger, ensure_ascii=False) + "\n当前由物理实例生成的BOM：" + json.dumps(compiled["bom"], ensure_ascii=False) + "\n由真实接触面生成的孔槽与连接件计划：" + json.dumps(manufacturing_plan, ensure_ascii=False) )}, ]}], "response_format": {"type": "json_object"}, "temperature": 0.0, "enable_thinking": True, "thinking_budget": 2048, "vl_high_resolution_images": True, "max_tokens": 6000, - }, "原图与结构图谱CAD复核失败", max_attempts=1, timeout_seconds=150, decoder=lambda text: json.loads(text)) + }, "原图与结构图谱CAD复核失败", max_attempts=2, timeout_seconds=150, + decoder=lambda text: normalize_visual_review( + json.loads(text), image_ids=image_ids, + dimension_ledger=dimension_ledger, + )) row["visual_review"] = review feedback = list(compiled["issues"]) + list(review.get("issues") or []) + list(review.get("uncertain") or []) if compiled["ok"] and manufacturing_plan.get("ok") and review.get("ok") is True and not feedback and not graph.get("uncertainties"): @@ -983,8 +1672,14 @@ def main(): report["status"] = "graph_not_passed" except Exception as exc: row["error"] = f"{type(exc).__name__}: {exc}" - report["status"] = "failed" - feedback = [{"code": "request_or_compile_error", "detail": row["error"]}] + dependency_feedback = terminal_dependency_feedback(exc) + if dependency_feedback: + row["terminal_dependency_error"] = dependency_feedback + report["status"] = "external_dependency_failed" + feedback = [dependency_feedback] + else: + report["status"] = "failed" + feedback = [{"code": "request_or_compile_error", "detail": row["error"]}] if not row.get("patch_error") and isinstance(row.get("compiled"), dict): candidate_blocker_count = round_blocker_count(row) row["end_to_end_blocker_count"] = candidate_blocker_count @@ -1022,9 +1717,21 @@ def main(): report["rounds"].append(row) (case_dir / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"case": case, "round": row["round"], "status": report["status"], "elapsed": row["elapsed_seconds"], "feedback": feedback[:8]}, ensure_ascii=False), flush=True) + if row.get("terminal_dependency_error"): + break if report["status"].startswith("candidate_visual_pass"): break - if row.get("patch_error") and has_minimum_axis_packing_dead_end(round_base_feedback): + resolved_scope_axes = {fact["axis"] for fact in scope_facts} + dead_end_axes = { + axis for item in round_base_feedback if isinstance(item, dict) + and item.get("minimum_separation_axis_infeasible") is True + for axis in (item.get("minimum_separation_axes") or []) + } + if ( + row.get("patch_error") + and has_minimum_axis_packing_dead_end(round_base_feedback) + and not (resolved_scope_axes & dead_end_axes) + ): report["status"] = "geometry_dead_end_requires_focused_source_adjudication" dead_end_feedback = [ item for item in round_base_feedback @@ -1108,6 +1815,21 @@ def main(): json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" ) break + if ( + row.get("patch_error") + and has_minimum_axis_packing_dead_end(round_base_feedback) + and resolved_scope_axes & dead_end_axes + ): + report["status"] = "verified_scope_fact_requires_additional_geometry_repair" + report["blocking"] = { + "code": "verified_scope_fact_not_yet_realized_in_geometry", + "resolved_axes": sorted(resolved_scope_axes & dead_end_axes), + "detail": "该尺寸口径已有可信答案，禁止再次询问；继续局部修复参考包络、实际占用包络和装配位置。", + "next_owner": "bounded_graph_repair_without_reasking_user", + } + (case_dir / "report.json").write_text( + json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8" + ) reports.append(report) (out / "report.json").write_text(json.dumps(reports, ensure_ascii=False, indent=2), encoding="utf-8") diff --git a/server_patch/ai_3d_model/ai-solid-wood-cost/structure_graph.py b/server_patch/ai_3d_model/ai-solid-wood-cost/structure_graph.py index b18b53f..28f3b81 100644 --- a/server_patch/ai_3d_model/ai-solid-wood-cost/structure_graph.py +++ b/server_patch/ai_3d_model/ai-solid-wood-cost/structure_graph.py @@ -830,12 +830,23 @@ def _validate_hardware_mountings( hardware_low = hardware["position_mm"][axis] hardware_high = hardware_low + hardware["size_mm"][axis] host_plane = host_low if face[1] == "-" else host_high + normal_overlap = max( + 0.0, min(host_high, hardware_high) - max(host_low, hardware_low) + ) if face[1] == "+": - actual_embed = max(0.0, host_plane - hardware_low) - gap = max(0.0, hardware_low - host_plane) + actual_embed = normal_overlap + gap = ( + hardware_low - host_plane if hardware_low > host_plane + else host_low - hardware_high if hardware_high < host_low + else max(0.0, host_plane - hardware_high) + ) else: - actual_embed = max(0.0, hardware_high - host_plane) - gap = max(0.0, host_plane - hardware_high) + actual_embed = normal_overlap + gap = ( + host_plane - hardware_high if hardware_high < host_plane + else hardware_low - host_high if hardware_low > host_high + else max(0.0, hardware_low - host_plane) + ) tangent_axes = [dimension for dimension in range(3) if dimension != axis] tangent_overlaps = [ min( @@ -895,13 +906,30 @@ def _validate_hardware_mountings( candidate_plane = ( candidate_host_low if candidate_face[1] == "-" else candidate_host_high ) + candidate_normal_overlap = max( + 0.0, + min(candidate_host_high, candidate_hardware_high) + - max(candidate_host_low, candidate_hardware_low), + ) if candidate_face[1] == "+": - candidate_embed = max(0.0, candidate_plane - candidate_hardware_low) - candidate_gap = max(0.0, candidate_hardware_low - candidate_plane) + candidate_embed = candidate_normal_overlap + candidate_gap = ( + candidate_hardware_low - candidate_plane + if candidate_hardware_low > candidate_plane + else candidate_host_low - candidate_hardware_high + if candidate_hardware_high < candidate_host_low + else max(0.0, candidate_plane - candidate_hardware_high) + ) outward_projection = max(0.0, candidate_hardware_high - candidate_plane) else: - candidate_embed = max(0.0, candidate_hardware_high - candidate_plane) - candidate_gap = max(0.0, candidate_plane - candidate_hardware_high) + candidate_embed = candidate_normal_overlap + candidate_gap = ( + candidate_plane - candidate_hardware_high + if candidate_hardware_high < candidate_plane + else candidate_hardware_low - candidate_host_high + if candidate_hardware_low > candidate_host_high + else max(0.0, candidate_hardware_low - candidate_plane) + ) outward_projection = max(0.0, candidate_plane - candidate_hardware_low) candidate_tangent_axes = [ dimension for dimension in range(3) if dimension != candidate_axis diff --git a/server_patch/ai_3d_model/ai-solid-wood-cost/test_manufacturing_oracle.py b/server_patch/ai_3d_model/ai-solid-wood-cost/test_manufacturing_oracle.py index b45cc00..f3f8a44 100644 --- a/server_patch/ai_3d_model/ai-solid-wood-cost/test_manufacturing_oracle.py +++ b/server_patch/ai_3d_model/ai-solid-wood-cost/test_manufacturing_oracle.py @@ -9,12 +9,17 @@ from unittest import mock from manufacturing_oracle import plan_manufacturing_features from probe_structure_graph import ( - authoritative_interference_feedback, cad_source, compile_lower_bound_can_improve, - dead_end_question_focus, deterministic_dead_end_question, + authoritative_interference_feedback, bind_scope_facts, cad_source, + compile_lower_bound_can_improve, dead_end_question_focus, + deterministic_dead_end_question, diagnostic_mounting_clearance_cuts, has_minimum_axis_packing_dead_end, - normalize_focused_source_adjudication, refresh_compiled_kernel_feedback, + enrich_hardware_feedback_with_verified_placements, + normalize_focused_source_adjudication, normalize_scope_facts, + normalize_visual_review, + propose_reference_scope_projection, refresh_compiled_kernel_feedback, round_blocker_count, select_resume_checkpoint, source_for_role, structured_kernel_feedback, structured_patch_feedback, + terminal_dependency_feedback, ) from structure_graph import SCHEMA_VERSION, compile_graph @@ -67,6 +72,92 @@ def compile_case(graph): class ManufacturingOracleTest(unittest.TestCase): + def test_missing_model_credentials_are_terminal_dependency_not_kernel_feedback(self): + feedback = terminal_dependency_feedback(RuntimeError("家具读图模型尚未配置密钥")) + self.assertEqual("model_credentials_missing", feedback["code"]) + self.assertFalse(feedback["retryable"]) + self.assertIsNone(terminal_dependency_feedback(RuntimeError("ordinary compile error"))) + + def test_visual_review_drops_self_declared_non_issues_and_unproven_proportion_guess(self): + result = normalize_visual_review({ + "ok": False, + "issues": [ + {"code": "COMPONENT_COUNT_MISMATCH", "component": "shelf", + "source_image_id": "image3", "evidence": "The count is correct.", + "correction": "No correction needed for shelf count."}, + {"code": "GEOMETRIC_PROPORTION_DISCREPANCY", "component": "leg", + "source_image_id": "image1", "evidence": "photo looks taller", + "correction": "guess a taller leg"}, + {"code": "DIMENSION_MISMATCH", "component": "leg", + "source_image_id": "image4", "evidence": "photo proportion differs", + "correction": "change the drawing without measured values"}, + ], + "uncertain": [], + }, image_ids={"image1", "image3", "image4"}, dimension_ledger={"z": {"value_mm": 800}}) + self.assertTrue(result["ok"]) + self.assertEqual([], result["issues"]) + self.assertEqual( + ["self_declared_non_issue", "unsupported_issue_code", "numeric_values_required"], + [row["reason"] for row in result["normalization_audit"]], + ) + + def test_visual_review_keeps_proven_numeric_dimension_mismatch(self): + result = normalize_visual_review({ + "ok": False, + "issues": [{ + "code": "numeric_dimension_mismatch", "component": "top", + "source_image_id": "image4", "evidence": "图纸宽1200，CAD宽1180", + "correction": "将顶板宽度改为1200mm", "expected_mm": 1200, "actual_mm": 1180, + }], + "uncertain": [], + }, image_ids={"image4"}, dimension_ledger={"x": {"value_mm": 1200}}) + self.assertFalse(result["ok"]) + self.assertEqual(1200.0, result["issues"][0]["expected_mm"]) + + def test_hardware_feedback_includes_atomic_compiler_verified_placement(self): + from test_structure_graph import anchored_door_with_hardware + + graph = anchored_door_with_hardware() + graph["envelope_mm"] = [700, 100, 700] + graph["parts"][0]["size_mm"][0] = 680 + graph["parts"][1]["size_mm"][1] = 50 + graph["instances"][1]["mountings"][0].update({ + "kind": "through", "embed_depth_mm": 18, + }) + graph["assemblies"][0]["bounds_mm"].update({"x": [0, 700], "y": [20, 100]}) + graph["parts"].append({ + "id": "right-side", "name": "侧板", "shape": "box", "functional_role": "side", + "thickness_axis": "x", "size_mm": [20, 80, 700], "observed_quantity": 1, + }) + graph["instances"].append({ + "id": "right-side-instance", "part_id": "right-side", "assembly_id": "moving-panel", + "placement": { + "x": {"mode": "absolute", "value_mm": 680}, + "y": {"mode": "absolute", "value_mm": 20}, + "z": {"mode": "absolute", "value_mm": 0}, + }, + }) + graph["instances"][1]["mountings"].append({ + "host_instance_id": "right-side-instance", "kind": "surface", "face": "x-", + "embed_depth_mm": 0, "evidence_ids": ["view"], + }) + compiled = compile_graph(graph, source_image_ids={"original-front"}) + issue = next( + row for row in compiled["issues"] + if row.get("code") == "hardware_mounting_geometry" + and row.get("host_instance_id") == "right-side-instance" + ) + enriched = enrich_hardware_feedback_with_verified_placements( + graph, compiled, [issue], + compile_candidate=lambda candidate: compile_graph( + candidate, source_image_ids={"original-front"} + ), + ) + candidates = enriched[0]["compatible_placement_patches"] + self.assertTrue(candidates) + self.assertEqual(650.0, candidates[0]["patches"][0]["value"]["value_mm"]) + self.assertEqual("full_graph_recompile_strict_issue_reduction", candidates[0]["validation"]) + def test_probe_resolves_semantic_role_filename_aliases(self): with tempfile.TemporaryDirectory() as folder: root = pathlib.Path(folder) @@ -137,6 +228,68 @@ class ManufacturingOracleTest(unittest.TestCase): self.assertFalse(plan["ok"]) self.assertIn("captured_interface_incompatible", {row["code"] for row in plan["issues"]}) + def test_shelf_edge_abutting_back_broad_face_does_not_invent_another_groove(self): + compiled = { + "ok": True, + "instances": [ + {"id": "shelf", "part_id": "shelf", "functional_role": "shelf", + "thickness_axis": "z", "material_family": "engineered_panel", + "position_mm": [10.0, 0.0, 50.0], "size_mm": [80.0, 40.0, 18.0], "axis_order": [0, 1, 2]}, + {"id": "back", "part_id": "back", "functional_role": "back", + "thickness_axis": "y", "material_family": "engineered_panel", + "position_mm": [10.0, 40.0, 0.0], "size_mm": [80.0, 5.0, 100.0], "axis_order": [0, 1, 2]}, + ], + "derived_contacts": [{ + "id": "shelf-back", "a": "shelf", "b": "back", + "face_a": "y+", "face_b": "y-", "interface_type": "edge_to_face", + "edge_instance_id": "shelf", "broad_face_instance_id": "back", + }], + "bom": [], + } + plan = plan_manufacturing_features(compiled) + self.assertTrue(plan["ok"], plan["issues"]) + self.assertEqual("non_structural_abutment", plan["connection_plans"][0]["method"]) + self.assertEqual(0, plan["cost_drivers"]["groove_operations"]) + self.assertEqual(0, plan["cost_drivers"]["hardware_instances"]) + + def test_orthogonal_capture_allowances_expand_receiver_groove_order_independently(self): + compiled = { + "ok": True, + "instances": [ + {"id": "top", "part_id": "top", "functional_role": "top", + "thickness_axis": "z", "material_family": "engineered_panel", + "position_mm": [0.0, 0.0, 100.0], "size_mm": [100.0, 50.0, 10.0], "axis_order": [0, 1, 2]}, + {"id": "left", "part_id": "side", "functional_role": "side", + "thickness_axis": "x", "material_family": "engineered_panel", + "position_mm": [0.0, 0.0, 10.0], "size_mm": [10.0, 50.0, 90.0], "axis_order": [0, 1, 2]}, + {"id": "right", "part_id": "side", "functional_role": "side", + "thickness_axis": "x", "material_family": "engineered_panel", + "position_mm": [90.0, 0.0, 10.0], "size_mm": [10.0, 50.0, 90.0], "axis_order": [0, 1, 2]}, + {"id": "back", "part_id": "back", "functional_role": "back", + "thickness_axis": "y", "material_family": "engineered_panel", + "position_mm": [10.0, 40.0, 10.0], "size_mm": [80.0, 5.0, 90.0], "axis_order": [0, 1, 2]}, + ], + "derived_contacts": [ + {"id": "top-back", "a": "top", "b": "back", "face_a": "z-", "face_b": "z+", + "interface_type": "edge_to_face", "edge_instance_id": "back", "broad_face_instance_id": "top"}, + {"id": "left-back", "a": "left", "b": "back", "face_a": "x+", "face_b": "x-", + "interface_type": "edge_to_face", "edge_instance_id": "back", "broad_face_instance_id": "left"}, + {"id": "right-back", "a": "right", "b": "back", "face_a": "x-", "face_b": "x+", + "interface_type": "edge_to_face", "edge_instance_id": "back", "broad_face_instance_id": "right"}, + ], + "bom": [], + } + plan = plan_manufacturing_features(compiled) + self.assertTrue(plan["ok"], plan["issues"]) + manufactured = plan["manufacturing_geometry_by_instance"]["back"] + self.assertEqual([6.0, 40.0, 10.0], manufactured["position_mm"]) + self.assertEqual([88.0, 5.0, 94.0], manufactured["size_mm"]) + top_groove = plan["features_by_instance"]["top"]["grooves"][0] + self.assertEqual(88.0, top_groove["world_size_mm"]["x"]) + self.assertEqual(5.5, top_groove["world_size_mm"]["y"]) + self.assertEqual(50.0, top_groove["world_center_mm"]["x"]) + self.assertEqual("final_insert_geometry_after_all_capture_allowances", top_groove["coverage_source"]) + def test_engineered_panel_mortise_is_blocked_by_material_not_display_name(self): compiled = compile_case(graph_for_role(role="top")) plan = plan_manufacturing_features(compiled, policy={"engineered_panel_joint": "mortise_tenon"}) @@ -301,6 +454,120 @@ class ManufacturingOracleTest(unittest.TestCase): self.assertIn("包含柜门", question["question_options"][0]) self.assertIn("不包含柜门", question["question_options"][1]) + def test_verified_user_scope_fact_is_bound_outside_model_patch_loop(self): + ledger = {"y": {"value_mm": 400, "source_image_ids": ["image4"]}} + facts = normalize_scope_facts({ + "id": "user-depth-scope", "source_type": "user", "axis": "y", + "ledger_scope": "reference", "confidence": 1.0, + "source_image_ids": ["image4"], + "text": "400mm只量柜体，不包含柜门厚度", + }, image_ids={"image1", "image4"}, dimension_ledger=ledger) + graph = graph_for_role("door") + graph["observations"].append({ + "id": "user-depth-scope", "source_type": "user", + "text": "模型伪造的相反答案", "confidence": 1.0, + }) + bound, user_evidence, adjudications, audit = bind_scope_facts(graph, facts) + observation = next(row for row in bound["observations"] if row["id"] == "user-depth-scope") + self.assertEqual("400mm只量柜体，不包含柜门厚度", observation["text"]) + self.assertEqual(observation, user_evidence["user-depth-scope"]) + self.assertEqual({}, adjudications) + self.assertEqual( + {"ledger_scope": "reference", "evidence_id": "user-depth-scope"}, + bound["dimension_scope_overrides"]["y"], + ) + self.assertEqual("authoritative_outside_model_patch_loop", audit[0]["binding"]) + + def test_independent_scope_fact_uses_separate_compiler_adjudication(self): + ledger = {"y": {"value_mm": 400, "source_image_ids": ["image4"]}} + facts = normalize_scope_facts({ + "id": "ai-depth-scope", "source_type": "independent_source_adjudication", + "axis": "y", "ledger_scope": "reference", "confidence": 0.91, + "source_image_ids": ["image2", "image4"], + "text": "尺寸箭头两端落在柜体而不是门板外表面", + }, image_ids={"image1", "image2", "image4"}, dimension_ledger=ledger) + bound, user_evidence, adjudications, _ = bind_scope_facts(graph_for_role("door"), facts) + self.assertEqual({}, user_evidence) + self.assertNotIn("ai-depth-scope", {row["id"] for row in bound["observations"]}) + self.assertEqual("ai-depth-scope", bound["dimension_scope_overrides"]["y"]["adjudication_id"]) + self.assertEqual("reference", adjudications["ai-depth-scope"]["ledger_scope"]) + + def test_scope_fact_rejects_low_confidence_or_missing_qualified_axis(self): + base = { + "id": "scope-y", "source_type": "independent_source_adjudication", + "axis": "y", "ledger_scope": "reference", "confidence": 0.7, + "source_image_ids": ["image4"], "text": "端点可能停在柜体", + } + with self.assertRaises(ValueError): + normalize_scope_facts( + base, image_ids={"image4"}, + dimension_ledger={"y": {"value_mm": 400}}, + ) + higher = dict(base, confidence=0.9, axis="x") + with self.assertRaises(ValueError): + normalize_scope_facts( + higher, image_ids={"image4"}, + dimension_ledger={"y": {"value_mm": 400}}, + ) + + def test_reference_scope_projection_moves_carcass_frame_but_not_front(self): + graph = { + "envelope_mm": [800, 400, 700], + "reference_bounds_mm": {"x": [0, 800], "y": [0, 400], "z": [0, 700]}, + "assemblies": [ + {"id": "carcass", "role": "carcass", "bounds_mm": { + "x": [0, 800], "y": [0, 400], "z": [0, 700], + }}, + {"id": "door-group", "role": "door", "bounds_mm": { + "x": [0, 800], "y": [0, 18], "z": [0, 700], + }}, + ], + "instances": [ + {"id": "side", "part_id": "side-part", "assembly_id": "carcass", "placement": { + "y": {"mode": "align", "self_anchor": "min", "target": "envelope", "target_anchor": "min"}, + }}, + {"id": "door", "part_id": "door-part", "assembly_id": "door-group", "placement": { + "y": {"mode": "align", "self_anchor": "min", "target": "envelope", "target_anchor": "min"}, + }}, + ], + } + compiled = { + "occupied_envelope_mm": [800, 400, 700], + "instances": [ + {"id": "side", "assembly_id": "carcass", "functional_role": "side", + "thickness_axis": "x", "position_mm": [0, 0, 0], "size_mm": [18, 400, 700]}, + {"id": "door", "assembly_id": "door-group", "functional_role": "door", + "thickness_axis": "y", "position_mm": [0, 0, 0], "size_mm": [800, 18, 700]}, + ], + } + projected, audit = propose_reference_scope_projection( + graph, compiled, [{ + "id": "scope-y", "axis": "y", "ledger_scope": "reference", + }], {"y": {"value_mm": 400}}, + ) + self.assertEqual([800, 418.0, 700], projected["envelope_mm"]) + self.assertEqual([18.0, 418.0], projected["reference_bounds_mm"]["y"]) + self.assertEqual([18.0, 418.0], projected["assemblies"][0]["bounds_mm"]["y"]) + self.assertEqual([0, 18], projected["assemblies"][1]["bounds_mm"]["y"]) + self.assertEqual("carcass", projected["instances"][0]["placement"]["y"]["target"]) + self.assertEqual("envelope", projected["instances"][1]["placement"]["y"]["target"]) + self.assertTrue(audit[0]["accepted"]) + + projected_again, second_audit = propose_reference_scope_projection( + projected, { + "occupied_envelope_mm": [800, 418, 700], + "instances": [ + {"id": "side", "assembly_id": "carcass", "functional_role": "side", + "thickness_axis": "x", "position_mm": [0, 18, 0], "size_mm": [18, 400, 700]}, + {"id": "door", "assembly_id": "door-group", "functional_role": "door", + "thickness_axis": "y", "position_mm": [0, 0, 0], "size_mm": [800, 18, 700]}, + ], + }, [{"id": "scope-y", "axis": "y", "ledger_scope": "reference"}], + {"y": {"value_mm": 400}}, + ) + self.assertIsNone(projected_again) + self.assertEqual("reference_scope_projection_already_bound", second_audit[0]["reason"]) + def test_resume_skips_rejected_round_and_chooses_latest_tie(self): rows = [ {"round": 1, "graph": {"id": "old"}, "compiled": {"issues": [{}]}}, diff --git a/server_patch/ai_3d_model/ai-solid-wood-cost/test_structure_graph.py b/server_patch/ai_3d_model/ai-solid-wood-cost/test_structure_graph.py index b0ebf95..c818fbc 100644 --- a/server_patch/ai_3d_model/ai-solid-wood-cost/test_structure_graph.py +++ b/server_patch/ai_3d_model/ai-solid-wood-cost/test_structure_graph.py @@ -161,6 +161,17 @@ class StructureGraphTest(unittest.TestCase): self.assertIn("disconnected_instance", codes) self.assertFalse(result["mounting_audit"]) + def test_hardware_entirely_beyond_host_inner_face_cannot_fake_through_mounting(self): + graph = anchored_door_with_hardware(kind="through", embed_depth_mm=18) + graph["instances"][1]["placement"]["y"]["value_mm"] = 50 + result = compile_case(graph) + problem = next( + row for row in result["issues"] if row["code"] == "hardware_mounting_geometry" + ) + self.assertGreater(problem["gap_mm"], 0) + self.assertEqual(0, problem["actual_embed_depth_mm"]) + self.assertFalse(result["mounting_audit"]) + def test_mortised_hardware_allows_only_declared_host_penetration(self): result = compile_case(anchored_door_with_hardware(kind="mortised", embed_depth_mm=5)) self.assertTrue(result["ok"], result["issues"]) @@ -192,7 +203,7 @@ class StructureGraphTest(unittest.TestCase): self.assertIn("厚度轴y的广面", problem["detail"]) self.assertTrue(problem["compatible_mounting_values"]) self.assertEqual( - {"y-", "y+"}, {row["face"] for row in problem["compatible_mounting_values"]} + {"y-"}, {row["face"] for row in problem["compatible_mounting_values"]} ) def test_role_axis_mode_accepts_direct_product_axes_without_instance_mapping(self):