feat: harden CAD resume and hardware validation

This commit is contained in:
chenpei 2026-09-08 01:42:41 +08:00
parent 695090d05a
commit 47ef7bbed9
6 changed files with 1143 additions and 23 deletions

View File

@ -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%仍未完成,接口状态必须保持测试失败/继续优化。

View File

@ -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():

View File

@ -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")

View File

@ -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

View File

@ -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": [{}]}},

View File

@ -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):