feat: halt infeasible CAD repair loops

This commit is contained in:
chenpei 2026-09-07 23:30:24 +08:00
parent 68766258b9
commit 809507820a
6 changed files with 578 additions and 10 deletions

View File

@ -78,6 +78,14 @@
- 续跑选择器排除任何带`patch_error`的拒绝轮次;编译阶段只有在当前问题数作为下界已经不可能优于上一端到端总分时才提前回滚,避免用旧编译问题数误杀可能显著减少B-rep冲突的候选。专项60项、完整196项回归通过。
- 当前仍缺少安装特征实体化:B-rep应依据安装模板在宿主生成孔、杯槽、沉孔或开口,同时把五金拆成可见体和嵌入体,不能永久靠允许碰撞对掩盖。下一阶段还要增加“参考尺寸包络/实际占用包络”语义,识别覆盖门、嵌入背板与净深尺寸,先做尺寸链可行性裁决再进入AI修复。
## v101 最小分离轴可行区间与防循环门禁(2026-09-07)
- 诊断B-rep会在已验证的埋入/贯穿安装宿主上减去五金嵌入包围盒,但这些减材显式标记为非生产特征。生产孔槽必须继续由供应商五金模板、连接面和实际规格生成,两者不可混用。
- 内核干涉复核始终针对刚生成的STEP绝对路径直接执行,并解析stdout JSON。引擎封装层的进度stderr、旧结果字段或进程退出码不再能覆盖真实碰撞对。
- 每个实体冲突会取重叠最小的分离轴,在零件所属总成或产品边界中构造允许的下坐标域,再逐个扣除与其它实体发生正体积相交的禁止区间。输出包含允许域、所有阻塞实体、禁止区间和剩余可行区间,可被人和AI独立复算。
- 若冲突两侧实体在最小分离轴上都没有任何可行区间,该问题标记`minimum_separation_axis_infeasible=true`。AI不得继续往返移件,只能依证据修正尺寸/外形口径/覆盖与内嵌语义,或建立显式槽口、避让等加工几何。
- 无用户参与的隔离探针在一轮原图修复未使端到端阻塞严格下降时立即停止,输出`geometry_dead_end_requires_focused_source_adjudication`。正式交互应先调用AI聚焦原图复读;只有高置信尺寸仍相互矛盾时,向用户集中询问一个能决定结构的问题,不重启全流程。
## 同行公开资料的可用结论
Zoo公开说明其代理会操作可编辑的参数化模型,执行几何代码,读取质量/体积等模型信息,并用多视图快照反复检查。我们借鉴“结构表示+几何执行反馈+视觉反馈”机制,不能据其宣传推断多张家具图片已经可以稳定得到生产级完整BOM,也不能据此承诺成本误差5%。

View File

@ -247,3 +247,13 @@
- v96还暴露两个未解决的真实边界:一是表装/埋入五金在结构编译器合法后,CAD实体仍是完整盒体,B-rep层尚未生成宿主安装孔/槽;二是样例深度链存在语义歧义,门厚20mm、层板深380mm、背板厚9mm与400mm总深无法在当前“所有实体必须包含于同一外包络”的假设下同时无穿插放置。必须先增加参考外形与实际外形/覆盖门语义或尺寸可行性裁决,不能靠移动板件伪造通过。
- 最终候选代码专项60项通过(0.063秒),完整回归196项通过(236.111秒);续跑会跳过所有带`patch_error`的拒绝轮次,编译早退仅在“编译问题下界已不可能优于已接受端到端总分”时触发。测试通过只证明防回退和既有功能未回退,不改变本次真实样例仍失败的结论。
- 本轮保持候选状态:五金挂载和修复事务更可靠,但生产级连接特征、真实多产品盲测以及冻结后成本误差≤5%均未完成,禁止标记成功或部署。
## 2026-09-07 v97-v101 安装碰撞去噪与尺寸可行性门禁(候选,未上线)
- v96第5轮图谱的B-rep原始复核有8个碰撞,其中6个是已经结构编译器验证的铰链/拉手埋入安装,不应指令AI把五金移离宿主。候选CAD构建现只从宿主减去这些已验证安装的诊断包围盒;其`production_feature=false`,不冒充真实铰杯孔、沉孔或孔位模板。
- 服务器隔离目录`mount-clearance-v97`实体检验证明:6个合法安装干涉被排除后,STEP仍保留18个实体且单体有效,内核只返回层板/背板1个真实冲突,体积208800mm³。这证明去噪没有吞掉无关结构错误。
- v97首次续跑暴露引擎封装层可能保留旧干涉结果,且相对STEP路径在切换`cwd`后被重复拼接。修复后对当前STEP直接执行`inspect interfere --format json`,优先解析stdout机器结果,现场稳定返回`['shelf_inst','back_panel_inst']`,不再降级为泛化内核错误。
- v100续跑复现了往返摆动:层板向前移开背板后立即与左右门板冲突,端到端阻塞仍为4,候选被事务门禁拒绝。根因是当前图纸同时给出总深400mm、门厚20mm、层板深380mm、背板厚9mm,在单一实体外包络假设下无法仅靠前后平移无冲突放置。
- 编译器新增精确单轴区间扣减:在所属总成/产品允许边界内,计算移动件不与所有切向相交实体碰撞的全部下坐标区间。该样例层板Y轴允许区间[0,20]被门板与背板完全覆盖;背板Y轴[0,391]也无可行子区间,因此返回`minimum_separation_axis_infeasible=true`,不再把局部位移候选误当成解。
- v101从同一已审计检查点隔离续跑,刷新基线为3个编译阻塞+1个真实内核碰撞。AI一轮修复产生11个编译问题,被拒绝后立即停止;最终状态`geometry_dead_end_requires_focused_source_adjudication`,仅执行1轮,没有再循环。下一交互节点必须先由AI聚焦复核外形口径、覆盖/内嵌关系、零件深度或明确避让加工;仍无法消解时只问用户一个聚焦问题。
- 最终完整回归205项全部通过(239.624秒)。以上现场运行均为`/tmp`隔离试验,每次启动前确认真实Chat API活动、等待、持久队列均为0;未覆盖生产文件、未重启生产服务。真实多产品盲测、生产级孔位模板和冻结后成本误差≤5%仍未完成。

View File

@ -13,6 +13,7 @@ import json
import os
import pathlib
import re
import subprocess
import time
from structure_graph import (
@ -30,6 +31,12 @@ ROLE_FILENAME_ALIASES = {
}
class AuthoritativeKernelError(RuntimeError):
def __init__(self, feedback: list[dict]):
super().__init__("CAD实体干涉检查未通过")
self.feedback = feedback
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]
@ -104,6 +111,67 @@ def structured_kernel_feedback(error: Exception | str) -> list[dict]:
}]
def authoritative_interference_feedback(ai3d, build_dir: pathlib.Path, step_path: str, built: dict | None = None) -> list[dict]:
"""Parse the kernel JSON itself; process exit behaviour is not a contract."""
step_file = pathlib.Path(step_path)
built_raw = str((built or {}).get("interference") or "").strip()
raw = ""
# The engine wrapper may retain an earlier validation payload in
# ``built['interference']`` even though the exported STEP is newer. When
# the STEP exists, inspect that exact artifact and make it authoritative.
if ai3d is not None and (step_file.is_file() or not built_raw):
python = str(ai3d.runtime_dir / "venv/bin/python")
inspect_tool = str(ai3d.skill_dir / "scripts/inspect")
try:
inspected = subprocess.run(
[
python, inspect_tool, "interfere",
str(step_file.resolve()) if step_file.is_file() else str(step_path),
"--format", "json",
],
cwd=build_dir, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
timeout=ai3d.timeout,
)
# The CLI writes progress to stderr and the machine result to
# stdout, including when it exits non-zero for a real clash.
raw = str(inspected.stdout or inspected.stderr or "").strip()
except Exception as exc:
raw = str(exc)
if not raw:
raw = built_raw
payload = _json_object_from_error(raw)
if isinstance(payload, dict) and payload.get("ok") is True:
return []
return structured_kernel_feedback(raw or "CAD内核未返回结构化干涉结果")
def build_and_authoritatively_validate(ai3d, build_dir: pathlib.Path, source_name: str, requested_exports=None) -> dict:
try:
built = ai3d._build_and_validate(
build_dir, source_name, requested_exports=list(requested_exports or [])
)
except Exception:
# Older engine wrappers raise from the interference command before
# returning its stdout JSON. If STEP exists, recover the authoritative
# pair list directly instead of degrading to a generic error.
step_path = build_dir / source_name[:-3]
if step_path.is_file():
feedback = authoritative_interference_feedback(
ai3d, build_dir, str(step_path), {}
)
if feedback and not (
len(feedback) == 1 and feedback[0].get("code") == "cad_kernel_error"
):
raise AuthoritativeKernelError(feedback)
raise
feedback = authoritative_interference_feedback(
ai3d, build_dir, str(built.get("step_path") or build_dir / source_name[:-3]), built
)
if feedback:
raise AuthoritativeKernelError(feedback)
return built
def round_blocker_count(row: dict) -> int:
"""Count end-to-end blockers for monotonic repair acceptance.
@ -135,6 +203,16 @@ def compile_lower_bound_can_improve(candidate_compile_issue_count: int, accepted
return candidate_compile_issue_count < accepted_total_blocker_count
def has_minimum_axis_packing_dead_end(feedback: list[dict]) -> bool:
"""Whether geometry proved that translation cannot repair a collision."""
return any(
isinstance(row, dict)
and row.get("code") == "solid_interference"
and row.get("minimum_separation_axis_infeasible") is True
for row in feedback
)
def select_resume_checkpoint(rows: list[dict]) -> dict:
"""Select only an accepted, internally consistent audited checkpoint."""
candidates = [
@ -266,6 +344,57 @@ def render_multiview(ai3d, build_dir: pathlib.Path, step_path: str) -> list[dict
return [{"camera": item["camera"], "path": item["path"]} for item in outputs if pathlib.Path(item["path"]).is_file()]
def diagnostic_mounting_clearance_cuts(compiled: dict, *, clearance_mm: float = 0.2) -> dict:
"""Create diagnostic-only host clearances for validated mounting overlap.
Hardware is currently represented by a conservative box envelope. A
mortised or through-mounted item therefore overlaps its host by design.
Subtracting only the already validated overlap keeps the B-rep collision
oracle from telling the repair model to move installed hardware away.
These boxes are never manufacturing drill or pocket instructions.
"""
instances = {str(row.get("id")): row for row in compiled.get("instances") or []}
cuts_by_host: dict[str, list[dict]] = {}
audit = []
pad = max(0.0, float(clearance_mm)) / 2.0
for mounting in compiled.get("mounting_audit") or []:
if not mounting.get("allows_host_interference"):
continue
hardware_id = str(mounting.get("instance_id") or "")
host_id = str(mounting.get("host_instance_id") or "")
hardware, host = instances.get(hardware_id), instances.get(host_id)
if not hardware or not host:
continue
low = [
max(float(hardware["position_mm"][axis]), float(host["position_mm"][axis]))
for axis in range(3)
]
high = [
min(
float(hardware["position_mm"][axis]) + float(hardware["size_mm"][axis]),
float(host["position_mm"][axis]) + float(host["size_mm"][axis]),
)
for axis in range(3)
]
if min(high[axis] - low[axis] for axis in range(3)) <= 0:
continue
size = [high[axis] - low[axis] + 2 * pad for axis in range(3)]
center = [(low[axis] + high[axis]) / 2.0 for axis in range(3)]
row = {
"host_instance_id": host_id,
"hardware_instance_id": hardware_id,
"kind": mounting.get("kind"),
"world_center_mm": {name: round(center[index], 6) for index, name in enumerate("xyz")},
"world_size_mm": {name: round(size[index], 6) for index, name in enumerate("xyz")},
"clearance_mm": round(float(clearance_mm), 6),
"source": "validated_mounting_overlap_envelope",
"production_feature": False,
}
cuts_by_host.setdefault(host_id, []).append(row)
audit.append(row)
return {"cuts_by_host": cuts_by_host, "audit": audit}
def cad_source(compiled, manufacturing_plan=None):
lines = [
"from build123d import Align, Axis, Box, Cylinder, Pos, Compound, Color",
@ -274,6 +403,7 @@ def cad_source(compiled, manufacturing_plan=None):
colors = [(0.72, 0.83, 0.96), (0.96, 0.77, 0.45), (0.8, 0.65, 0.86), (0.5, 0.82, 0.67)]
feature_map = (manufacturing_plan or {}).get("features_by_instance") or {}
manufacturing_geometry = (manufacturing_plan or {}).get("manufacturing_geometry_by_instance") or {}
mounting_clearances = diagnostic_mounting_clearance_cuts(compiled)["cuts_by_host"]
for index, inst in enumerate(compiled["instances"]):
adjusted = manufacturing_geometry.get(str(inst["id"])) or {}
size = adjusted.get("size_mm") or inst["size_mm"]
@ -300,6 +430,13 @@ def cad_source(compiled, manufacturing_plan=None):
f" groove_{index}_{feature_index} = Box({groove_size['x']!r}, {groove_size['y']!r}, {groove_size['z']!r}, align=(Align.CENTER, Align.CENTER, Align.CENTER)).moved(Pos({world['x']!r}, {world['y']!r}, {world['z']!r}))",
f" part = part - groove_{index}_{feature_index}",
])
for cut_index, cut in enumerate(mounting_clearances.get(str(inst["id"])) or [], 1):
world, cut_size = cut["world_center_mm"], cut["world_size_mm"]
lines.extend([
" # Diagnostic-only clearance for a validated mounted-hardware envelope.",
f" mount_clearance_{index}_{cut_index} = Box({cut_size['x']!r}, {cut_size['y']!r}, {cut_size['z']!r}, align=(Align.CENTER, Align.CENTER, Align.CENTER)).moved(Pos({world['x']!r}, {world['y']!r}, {world['z']!r}))",
f" part = part - mount_clearance_{index}_{cut_index}",
])
lines.extend([
f" part.label = {str(inst['id'])!r}",
f" part.color = Color{colors[index % len(colors)]!r}", " parts.append(part)",
@ -308,6 +445,30 @@ def cad_source(compiled, manufacturing_plan=None):
return "\n".join(lines) + "\n"
def refresh_compiled_kernel_feedback(ai3d, build_dir: pathlib.Path, compiled: dict, manufacturing_plan: dict) -> dict:
"""Rebuild one audited graph so resumed repair never consumes stale CAD feedback."""
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"))
try:
built = build_and_authoritatively_validate(ai3d, build_dir, source.name)
except Exception as exc:
return {
"feedback": (
exc.feedback if isinstance(exc, AuthoritativeKernelError)
else structured_kernel_feedback(exc)
),
"error": f"{type(exc).__name__}: {exc}",
"diagnostic_mounting_clearance_audit": diagnostic_mounting_clearance_cuts(compiled)["audit"],
}
return {
"feedback": [],
"built": built,
"diagnostic_mounting_clearance_audit": diagnostic_mounting_clearance_cuts(compiled)["audit"],
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--engine", required=True)
@ -463,14 +624,28 @@ def main():
topology_contract=topology_contract,
)
accepted_compile_issue_count = len(refreshed_compiled.get("issues") or [])
prior_kernel_feedback = list(best.get("cad_kernel_feedback") or [])
refreshed_manufacturing_plan = plan_manufacturing_features(
refreshed_compiled,
policy={"default_material_family": args.material_family},
)
refreshed_kernel = refresh_compiled_kernel_feedback(
ai3d, case_dir / "resume-kernel-refresh", refreshed_compiled,
refreshed_manufacturing_plan,
)
prior_kernel_feedback = list(refreshed_kernel.get("feedback") or [])
refreshed_checkpoint = {
"compiled": refreshed_compiled,
"manufacturing_plan": refreshed_manufacturing_plan,
"cad_kernel_feedback": prior_kernel_feedback,
"visual_review": best.get("visual_review"),
}
accepted_total_blocker_count = round_blocker_count(refreshed_checkpoint)
feedback = (
list(refreshed_compiled.get("issues") or [])
+ (
list(refreshed_manufacturing_plan.get("issues") or [])
if refreshed_compiled.get("ok") else []
)
+ prior_kernel_feedback
+ list((best.get("visual_review") or {}).get("issues") or [])
+ list((best.get("visual_review") or {}).get("uncertain") or [])
@ -482,6 +657,11 @@ def main():
"source_end_to_end_blocker_count": accepted_total_blocker_count,
"selection": "minimum_end_to_end_blocker_count_then_latest_round",
"compiler_feedback_refreshed": True,
"kernel_feedback_refreshed": True,
"refreshed_kernel_blocker_count": len(prior_kernel_feedback),
"diagnostic_mounting_clearance_count": len(
refreshed_kernel.get("diagnostic_mounting_clearance_audit") or []
),
}
for turn in range(max(1, min(args.rounds, 5))):
started = time.time()
@ -514,6 +694,10 @@ def main():
"最小必要位移且符合真实装配方向的候选;位移量相同时再选"
"remaining_collision_count更少的候选,逐字采用其patches;"
"候选中总成bounds修改与唯一结构件位移是一个原子修复,不得只做一半。"
"若minimum_separation_axis_infeasible=true,几何已证明最短分离方向内没有可用位置;"
"不得再把同一零件在两个障碍物之间往返移动。必须回看原图,只在有证据时修正"
"部件尺寸、外形口径、覆盖/内嵌语义或显式槽口;若原图高置信尺寸本身互相矛盾,"
"在uncertainties中记录准确冲突和唯一需确认的问题,不得猜一个数值强行通过。"
"修正定位时优先使用稳定实例ID。若某轴当前没有value_mm或offset_mm,不得replace不存在的叶子字段;"
"必须replace整个轴对象,或对align轴用add新增offset_mm。"
"合法示例:{\"patches\":[{\"op\":\"replace\",\"path\":\"/instances/door_left_inst/placement/x\","
@ -604,13 +788,19 @@ def main():
build_dir = case_dir / f"round-{row['round']}"
build_dir.mkdir(exist_ok=True)
source = build_dir / "assembly.step.py"
row["diagnostic_mounting_clearance_audit"] = diagnostic_mounting_clearance_cuts(compiled)["audit"]
source.write_text(cad_source(compiled, manufacturing_plan), encoding="utf-8")
ai3d._validate_source(source.read_text())
try:
built = ai3d._build_and_validate(build_dir, source.name, requested_exports=["glb"])
built = build_and_authoritatively_validate(
ai3d, build_dir, source.name, requested_exports=["glb"]
)
except Exception as exc:
row["cad_kernel_error"] = f"{type(exc).__name__}: {exc}"
kernel_feedback = structured_kernel_feedback(exc)
kernel_feedback = (
exc.feedback if isinstance(exc, AuthoritativeKernelError)
else structured_kernel_feedback(exc)
)
row["cad_kernel_feedback"] = kernel_feedback
feedback += kernel_feedback
report["status"] = (
@ -698,6 +888,21 @@ def main():
print(json.dumps({"case": case, "round": row["round"], "status": report["status"], "elapsed": row["elapsed_seconds"], "feedback": feedback[:8]}, ensure_ascii=False), flush=True)
if report["status"].startswith("candidate_visual_pass"):
break
if row.get("patch_error") and has_minimum_axis_packing_dead_end(round_base_feedback):
report["status"] = "geometry_dead_end_requires_focused_source_adjudication"
report["blocking"] = {
"code": "minimum_axis_packing_dead_end",
"detail": (
"最短分离轴没有合法放置区间,且一轮原图修复未使端到端阻塞下降。"
"已停止前后循环移动;下一步必须聚焦判定外形口径、覆盖/内嵌关系、"
"部件尺寸或明确避让加工。"
),
"next_owner": "ai_focused_source_adjudication_then_single_user_question_if_still_ambiguous",
}
(case_dir / "report.json").write_text(
json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
)
break
reports.append(report)
(out / "report.json").write_text(json.dumps(reports, ensure_ascii=False, indent=2), encoding="utf-8")

View File

@ -585,6 +585,96 @@ def _aabb_interference(a: dict, b: dict, *, tolerance_mm: float = 0.6) -> dict |
}
def _axis_packing_feasibility(
moving: dict,
instance_rows: list[dict],
bounds: list[list[float]],
*,
axes: list[int] | None = None,
ignored_pairs: set[frozenset] | None = None,
tolerance_mm: float = 0.6,
) -> list[dict]:
"""Find every collision-free lower-coordinate interval on selected axes.
The other two coordinates remain fixed. This is a diagnostic, not an
automatic layout solver: it proves when repeated front/back (or any-axis)
shuffling cannot fit a solid inside its permitted envelope. The caller
still decides whether the source dimension, envelope semantics, or an
explicit machining feature must change.
"""
ignored_pairs = ignored_pairs or set()
selected_axes = list(range(3)) if axes is None else list(axes)
answer = []
for axis in selected_axes:
axis_name = "xyz"[axis]
size = float(moving["size_mm"][axis])
domain_low = float(bounds[axis][0])
domain_high = float(bounds[axis][1]) - size
blockers = []
forbidden = []
if domain_high >= domain_low - tolerance_mm:
for other in instance_rows:
if other["id"] == moving["id"]:
continue
if frozenset((moving["id"], other["id"])) in ignored_pairs:
continue
tangent_axes = [value for value in range(3) if value != axis]
tangent_overlap = [
min(
float(moving["position_mm"][value]) + float(moving["size_mm"][value]),
float(other["position_mm"][value]) + float(other["size_mm"][value]),
) - max(
float(moving["position_mm"][value]),
float(other["position_mm"][value]),
)
for value in tangent_axes
]
if min(tangent_overlap) <= tolerance_mm:
continue
other_low = float(other["position_mm"][axis])
other_high = other_low + float(other["size_mm"][axis])
# Positive-volume overlap occurs strictly inside this interval
# of possible lower coordinates for ``moving``.
forbidden_low = other_low + tolerance_mm - size
forbidden_high = other_high - tolerance_mm
if forbidden_high <= domain_low or forbidden_low >= domain_high:
continue
clipped = [
max(domain_low, forbidden_low),
min(domain_high, forbidden_high),
]
forbidden.append(clipped)
blockers.append({
"instance_id": other["id"],
"forbidden_lower_position_mm": [round(value, 6) for value in clipped],
"tangent_overlap_mm": [round(float(value), 6) for value in tangent_overlap],
})
feasible = [[domain_low, domain_high]] if domain_high >= domain_low else []
for forbidden_low, forbidden_high in sorted(forbidden):
remaining = []
for segment_low, segment_high in feasible:
if forbidden_high <= segment_low or forbidden_low >= segment_high:
remaining.append([segment_low, segment_high])
continue
if forbidden_low > segment_low:
remaining.append([segment_low, min(segment_high, forbidden_low)])
if forbidden_high < segment_high:
remaining.append([max(segment_low, forbidden_high), segment_high])
feasible = remaining
answer.append({
"moving_instance_id": moving["id"],
"axis": axis_name,
"allowed_lower_position_domain_mm": [round(domain_low, 6), round(domain_high, 6)],
"feasible_lower_position_intervals_mm": [
[round(value, 6) for value in interval] for interval in feasible
],
"blocking_instances": blockers,
"feasible_without_dimension_or_bounds_change": bool(feasible),
"source": "one_axis_exact_aabb_interval_subtraction",
})
return answer
def _derive_aabb_contacts(instances: dict, *, tolerance_mm: float = 0.6) -> list[dict]:
"""Derive real positive-area contacts inside each declared assembly.
@ -1184,6 +1274,10 @@ def compile_graph(
frozenset((row["instance_id"], row["host_instance_id"])): row
for row in mounting_audit
}
ignored_interference_pairs = {
pair for pair, row in valid_mount_pairs.items()
if row.get("allows_host_interference")
}
valid_same_assembly_hardware = {
row["instance_id"] for row in mounting_audit if row["same_assembly"]
}
@ -1611,7 +1705,11 @@ def compile_graph(
simulated["position_mm"][axis] = new_low
remaining = sum(
1 for other in instance_rows
if other["id"] != moving_id and _aabb_interference(simulated, other)
if (
other["id"] != moving_id
and frozenset((moving_id, other["id"])) not in ignored_interference_pairs
and _aabb_interference(simulated, other)
)
)
patches = [{
"op": "replace",
@ -1647,13 +1745,54 @@ def compile_graph(
collision["mounting_kind"] = mounting["kind"]
collision["declared_embed_depth_mm"] = mounting["declared_embed_depth_mm"]
continue
minimum_overlap = min(float(value) for value in collision["overlap_mm"])
minimum_axes = [
axis for axis, value in enumerate(collision["overlap_mm"])
if abs(float(value) - minimum_overlap) <= 1e-6
]
packing_feasibility = []
for moving in (a, b):
assembly_id = moving.get("assembly_id")
structural_rows = structural_members_by_assembly.get(assembly_id) or []
permitted_bounds = assembly_bounds.get(assembly_id)
if (
envelope
and len(structural_rows) == 1
and structural_rows[0]["id"] == moving["id"]
):
permitted_bounds = [[0.0, float(value)] for value in envelope]
if not permitted_bounds and envelope:
permitted_bounds = [[0.0, float(value)] for value in envelope]
if permitted_bounds:
packing_feasibility.extend(_axis_packing_feasibility(
moving, instance_rows, permitted_bounds,
axes=minimum_axes,
ignored_pairs=ignored_interference_pairs,
))
minimum_axis_infeasible = bool(packing_feasibility) and all(
not row["feasible_without_dimension_or_bounds_change"]
for row in packing_feasibility
)
issue(
"solid_interference", f"/instances/{b['id']}",
f"实体{a['id']}与{b['id']}发生正体积穿插,不能作为可制造装配",
compatible_separation_candidates=collision_separation_candidates(a, b),
minimum_separation_axes=["xyz"[axis] for axis in minimum_axes],
minimum_axis_packing_feasibility=packing_feasibility,
minimum_separation_axis_infeasible=minimum_axis_infeasible,
repair_scope=(
"source_dimension_or_envelope_semantics_or_explicit_machining_feature"
if minimum_axis_infeasible else "placement_or_dimension"
),
correction=(
"必须回看原图选择compatible_separation_candidates中与真实结构一致的最小位移;"
"候选只保证几何可行,不代替图像证据。"
(
"最小分离轴在所属边界内没有任何无碰撞位置;禁止在前后障碍之间循环平移。"
"必须回看原图核对外形尺寸口径、覆盖/内嵌关系、零件尺寸,或显式建模槽口/避让;"
"若高置信原图尺寸互相矛盾,必须输出单一聚焦的尺寸确认问题,不得臆造修改。"
if minimum_axis_infeasible else
"必须回看原图选择compatible_separation_candidates中与真实结构一致的最小位移;"
"候选只保证几何可行,不代替图像证据。"
)
),
**collision,
)

View File

@ -1,15 +1,19 @@
"""General manufacturing-feature tests; no SKU/name answer constants."""
import ast
import copy
import os
import pathlib
import tempfile
import unittest
from unittest import mock
from manufacturing_oracle import plan_manufacturing_features
from probe_structure_graph import (
cad_source, compile_lower_bound_can_improve, round_blocker_count,
select_resume_checkpoint, source_for_role, structured_kernel_feedback,
structured_patch_feedback,
authoritative_interference_feedback, cad_source, compile_lower_bound_can_improve,
diagnostic_mounting_clearance_cuts, has_minimum_axis_packing_dead_end,
refresh_compiled_kernel_feedback,
round_blocker_count, select_resume_checkpoint, source_for_role,
structured_kernel_feedback, structured_patch_feedback,
)
from structure_graph import SCHEMA_VERSION, compile_graph
@ -212,6 +216,15 @@ class ManufacturingOracleTest(unittest.TestCase):
self.assertFalse(compile_lower_bound_can_improve(10, 10))
self.assertFalse(compile_lower_bound_can_improve(11, 10))
def test_packing_dead_end_gate_ignores_ordinary_collisions(self):
self.assertFalse(has_minimum_axis_packing_dead_end([
{"code": "solid_interference", "minimum_separation_axis_infeasible": False},
{"code": "hardware_mounting_geometry"},
]))
self.assertTrue(has_minimum_axis_packing_dead_end([
{"code": "solid_interference", "minimum_separation_axis_infeasible": True},
]))
def test_resume_skips_rejected_round_and_chooses_latest_tie(self):
rows = [
{"round": 1, "graph": {"id": "old"}, "compiled": {"issues": [{}]}},
@ -224,6 +237,169 @@ class ManufacturingOracleTest(unittest.TestCase):
]
self.assertEqual("latest", select_resume_checkpoint(rows)["graph"]["id"])
def test_diagnostic_mounting_clearance_is_limited_to_validated_penetration(self):
compiled = {
"instances": [
{"id": "door", "position_mm": [0, 20, 0], "size_mm": [600, 18, 700]},
{"id": "hinge", "position_mm": [10, 15, 100], "size_mm": [30, 10, 20]},
{"id": "handle", "position_mm": [300, 10, 300], "size_mm": [20, 10, 30]},
],
"mounting_audit": [
{"instance_id": "hinge", "host_instance_id": "door", "kind": "mortised",
"allows_host_interference": True},
{"instance_id": "handle", "host_instance_id": "door", "kind": "surface",
"allows_host_interference": False},
],
}
result = diagnostic_mounting_clearance_cuts(compiled)
self.assertEqual(1, len(result["audit"]))
cut = result["audit"][0]
self.assertEqual("hinge", cut["hardware_instance_id"])
self.assertEqual({"x": 30.2, "y": 5.2, "z": 20.2}, cut["world_size_mm"])
self.assertFalse(cut["production_feature"])
def test_cad_source_subtracts_only_audited_mounting_clearance(self):
compiled = {
"instances": [
{"id": "door", "position_mm": [0, 20, 0], "size_mm": [600, 18, 700]},
{"id": "hinge", "position_mm": [10, 15, 100], "size_mm": [30, 10, 20]},
],
"mounting_audit": [{
"instance_id": "hinge", "host_instance_id": "door", "kind": "mortised",
"allows_host_interference": True,
}],
}
source = cad_source(compiled)
self.assertIn("mount_clearance_0_1", source)
self.assertIn("part = part - mount_clearance_0_1", source)
ast.parse(source)
def test_resume_kernel_refresh_rebuilds_current_graph_feedback(self):
compiled = {
"instances": [{"id": "panel", "position_mm": [0, 0, 0], "size_mm": [10, 20, 30]}],
"mounting_audit": [],
}
class FakeEngine:
def __init__(self, error=None):
self.error = error
self.validated = False
def _validate_source(self, source):
ast.parse(source)
self.validated = True
def _build_and_validate(self, build_dir, source_name, requested_exports):
if self.error:
raise self.error
return {
"step_path": str(build_dir / "assembly.step"),
"interference": '{"ok":true,"clashes":[],"errors":[]}',
}
with tempfile.TemporaryDirectory() as directory:
engine = FakeEngine(RuntimeError('{"ok":false,"clashes":[],"errors":["current failure"]}'))
failed = refresh_compiled_kernel_feedback(
engine, pathlib.Path(directory), compiled, {}
)
self.assertTrue(engine.validated)
self.assertEqual("cad_kernel_validation_error", failed["feedback"][0]["code"])
with tempfile.TemporaryDirectory() as directory:
passed = refresh_compiled_kernel_feedback(
FakeEngine(), pathlib.Path(directory), compiled, {}
)
self.assertEqual([], passed["feedback"])
self.assertIn("built", passed)
def test_authoritative_kernel_parser_does_not_trust_successful_process_alone(self):
built = {
"interference": (
'{"ok":false,"clashes":[{"a":{"name":"shelf"},'
'"b":{"name":"back"},"volume":12,"bounds":{"min":[0,0,0],'
'"max":[1,2,6]}}],"errors":[]}'
)
}
feedback = authoritative_interference_feedback(None, pathlib.Path("."), "unused.step", built)
self.assertEqual(1, len(feedback))
self.assertEqual(["shelf", "back"], feedback[0]["instance_ids"])
def test_authoritative_kernel_parser_prefers_existing_step_over_stale_wrapper_payload(self):
class FakeEngine:
runtime_dir = pathlib.Path("/runtime")
skill_dir = pathlib.Path("/skill")
timeout = 30
stale = {
"interference": '{"ok":false,"clashes":[],"errors":["stale wrapper error"]}'
}
completed = __import__("subprocess").CompletedProcess(
[], 2,
stdout=(
'{"ok":false,"clashes":[{"a":{"name":"shelf"},'
'"b":{"name":"back"},"volume":12,"bounds":{"min":[0,0,0],'
'"max":[1,2,6]}}],"errors":[]}'
),
stderr="",
)
with tempfile.TemporaryDirectory() as directory:
step = pathlib.Path(directory) / "assembly.step"
step.touch()
with mock.patch("probe_structure_graph.subprocess.run", return_value=completed):
feedback = authoritative_interference_feedback(
FakeEngine(), pathlib.Path(directory), str(step), stale
)
self.assertEqual(["shelf", "back"], feedback[0]["instance_ids"])
def test_authoritative_kernel_parser_resolves_relative_step_before_changing_cwd(self):
class FakeEngine:
runtime_dir = pathlib.Path("/runtime")
skill_dir = pathlib.Path("/skill")
timeout = 30
completed = __import__("subprocess").CompletedProcess(
[], 0, stdout='{"ok":true,"clashes":[],"errors":[]}', stderr=""
)
with tempfile.TemporaryDirectory() as directory:
root = pathlib.Path(directory)
build_dir = root / "relative-build"
build_dir.mkdir()
step = build_dir / "assembly.step"
step.touch()
old_cwd = pathlib.Path.cwd()
try:
os.chdir(root)
with mock.patch("probe_structure_graph.subprocess.run", return_value=completed) as run:
feedback = authoritative_interference_feedback(
FakeEngine(), pathlib.Path("relative-build"),
"relative-build/assembly.step", {}
)
finally:
os.chdir(old_cwd)
self.assertEqual([], feedback)
self.assertTrue(pathlib.Path(run.call_args.args[0][3]).is_absolute())
def test_authoritative_kernel_parser_prefers_stdout_json_over_stderr_progress(self):
class FakeEngine:
runtime_dir = pathlib.Path("/runtime")
skill_dir = pathlib.Path("/skill")
timeout = 30
completed = __import__("subprocess").CompletedProcess(
[], 2,
stdout=(
'{"ok":false,"clashes":[{"a":{"name":"shelf"},'
'"b":{"name":"back"},"volume":12,"bounds":{"min":[0,0,0],'
'"max":[1,2,6]}}],"errors":[]}'
),
stderr="building geometry...",
)
with mock.patch("probe_structure_graph.subprocess.run", return_value=completed):
feedback = authoritative_interference_feedback(
FakeEngine(), pathlib.Path("."), "assembly.step", {}
)
self.assertEqual(["shelf", "back"], feedback[0]["instance_ids"])
def test_rejected_patch_feedback_uses_stable_id_and_existing_axis_shape(self):
graph = graph_for_role()
bad_patch = [{

View File

@ -4,7 +4,7 @@ import random
import unittest
from structure_graph import (
GraphError, ParameterResolver, SCHEMA_VERSION, apply_graph_patch,
GraphError, ParameterResolver, SCHEMA_VERSION, _axis_packing_feasibility, apply_graph_patch,
build_dimension_ledger, build_source_topology_contract, compile_graph,
normalize_graph_patch_payload,
)
@ -704,6 +704,36 @@ class StructureGraphTest(unittest.TestCase):
[row["path"] for row in candidate["patches"]],
)
def test_axis_packing_proves_when_front_member_shelf_and_back_cannot_fit(self):
shelf = {
"id": "moving", "position_mm": [0, 20, 40],
"size_mm": [100, 380, 20],
}
rows = [
shelf,
{"id": "front-left", "position_mm": [0, 0, 0], "size_mm": [50, 20, 100]},
{"id": "front-right", "position_mm": [50, 0, 0], "size_mm": [50, 20, 100]},
{"id": "back", "position_mm": [0, 382, 0], "size_mm": [100, 9, 100]},
]
result = _axis_packing_feasibility(
shelf, rows, [[0, 100], [0, 400], [0, 100]], axes=[1]
)[0]
self.assertFalse(result["feasible_without_dimension_or_bounds_change"])
self.assertEqual([], result["feasible_lower_position_intervals_mm"])
self.assertEqual(
{"front-left", "front-right", "back"},
{row["instance_id"] for row in result["blocking_instances"]},
)
shorter = copy.deepcopy(shelf)
shorter["size_mm"][1] = 360
rows[0] = shorter
feasible = _axis_packing_feasibility(
shorter, rows, [[0, 100], [0, 400], [0, 100]], axes=[1]
)[0]
self.assertTrue(feasible["feasible_without_dimension_or_bounds_change"])
self.assertTrue(feasible["feasible_lower_position_intervals_mm"])
def test_anchored_duplicate_instance_id_and_loose_assembly_bounds_are_blocking(self):
duplicate = anchored_frame()
duplicate["instances"].append(copy.deepcopy(duplicate["instances"][0]))