fix: partition initial CAD graph generation

server_patch/ai_3d_model/ai-solid-wood-cost/VALIDATION_LEDGER.md

@@ -325,3 +325,12 @@
 - 真实响应在结构事实 JSON 第 193 行附近未闭合，尺寸专项响应返回 2300 tokens 后同样未闭合。候选请求层将其归类为 `model_output_truncated`，责任层为 `model_gateway_capacity`、`retryable=false`；报告 `rounds=[]`，证明没有再进入图谱轮次或重复调用。
 - 该现场结果验证了“截断即终止、不循环、不误生成 CAD/BOM”的通用修复，但也证明当前大而全的结构事实提示仍超过模型有效输出边界。下一步不再提升原样重试次数，而是把多视图识别拆成装配概览、柜体板件、门抽/五金和尺寸账本等小型独立事实，再由确定性参数化编译器合并并执行跨事实一致性门禁。
 - 本次仍未生成可验收 CAD/BOM，未部署候选；不同公司产品冻结后盲测、部件/孔槽/五金完整性以及总成本误差不超过 5% 均未完成，测试状态继续保持失败。
+
+## 2026-09-08 v122-v123 分阶段图谱生成与局部定位失败隔离（候选，未上线）
+
+- 新阿里云凭据以受限文件方式更新后，使用 `qwen3-vl-plus` 的真实最小请求返回 HTTP 200；生产服务没有重启。随后在服务器 `/tmp/dalongxia-cad-v85-eI8HeF` 继续四张原始图隔离试验，未读取公司 BOM，报告明确记录 `company_bom_used_in_inference=false`。
+- v122 的源证据资格审查通过，整体尺寸固定为 1200×400×800mm，模型不再发生请求超时：首轮图谱约 118.42 秒返回，第二轮局部修复约 90.94 秒返回。但两轮编译分别有 37、35 项问题且物理实例均为 0，最终 `graph_not_passed`，不能作为成功模型。
+- 现场报告确认首个真实错误是门总成实例把 `carcass` 当作跨总成定位目标。旧定位求解器遇到这一个非法引用后会丢弃全图所有已合法求解实例，进而伪造空总成、无实例零件、数量冲突和源拓扑冲突等数十项级联问题。该全局失败行为已停止使用。
+- 定位求解现按实例依赖树隔离失败：非法实例及依赖它的实例返回稳定 ID 路径，其余合法坐标、审计和 BOM 实例继续保留；对已知定位失败的零件、角色和总成暂缓派生数量/边界级联检查。新增测试证明一个跨总成引用只阻断受影响实例，不再把 `/instances` 整体清空。
+- 初始图谱由一次大请求改为四个有边界的多模态阶段：核心证据/参数/总成/零件、非五金物理实例、五金模板安装、总成运动关系。每一段都重新获得全部原图；程序在合并前强制校验稳定 ID、逐零件实例数量、三轴定位结构、五金宿主和总成连接端点。初始结构实例只允许锚定整体或自身总成，禁止跨总成和实例依赖；失败阶段会保存审计并停止合并，不把半份图谱送入 CAD。
+- 本地完整回归 236 项通过（238.562 秒），其中 1 项真实 OCCT 内核测试按设计仅在服务器开关启用时跳过；Python 语法检查通过。分阶段通用算法尚未做新的服务器原图实跑，也未部署生产；不同公司产品冻结后 BOM/成本盲测与总成本误差不超过 5% 仍未完成。

server_patch/ai_3d_model/ai-solid-wood-cost/probe_structure_graph.py

@@ -39,6 +39,14 @@ class AuthoritativeKernelError(RuntimeError):
         self.feedback = feedback
 
 
+class PartitionedGraphError(GraphError):
+    """A bounded generation pass failed before an unsafe graph was merged."""
+
+    def __init__(self, message: str, audit: list[dict]):
+        super().__init__(message)
+        self.audit = copy.deepcopy(audit)
+
+
 def terminal_dependency_feedback(error: Exception | str) -> dict | None:
     """Classify non-geometric configuration failures that retries cannot fix."""
     message = str(error or "").strip()
@@ -66,6 +74,14 @@ def terminal_dependency_feedback(error: Exception | str) -> dict | None:
             "next_owner": "model_gateway_capacity",
             "retryable": False,
         }
+    if any(token in message.lower() for token in ("read operation timed out", "timed out", "timeout")):
+        return {
+            "code": "model_request_timeout",
+            "severity": "blocking",
+            "detail": "模型请求在当前结构任务时间窗内未返回；禁止原样重复同一大请求",
+            "next_owner": "request_partitioning_or_model_routing",
+            "retryable": False,
+        }
     return None
 
 
@@ -881,6 +897,269 @@ mounting_center_mm是孔型中心，必须在宿主安装平面上且完整孔
 输出紧凑有效JSON，最多40个parts、100个instances，不输出油漆材料、成本或孔位模板。"""
 
 
+def generate_partitioned_initial_graph(
+    *, engine, model: str, images: list[dict], evidence_context: str,
+    hardware_contract: dict,
+) -> tuple[dict, list[dict]]:
+    """Generate one graph through bounded, independently validated AI passes.
+
+    The source images are shown to every semantic pass.  Deterministic merge
+    gates reject cross-assembly placement references, missing physical
+    instances and invalid motion endpoints before the compiler sees them.
+    Company BOM data is deliberately absent from this interface.
+    """
+    audit = []
+
+    def request(stage: str, instruction: str, *, max_tokens: int, timeout_seconds: int):
+        row = {
+            "stage": stage, "status": "running", "max_tokens": max_tokens,
+            "timeout_seconds": timeout_seconds, "source_images_included": True,
+            "company_bom_used_in_inference": False,
+        }
+        audit.append(row)
+        started = time.time()
+        try:
+            result = engine._request_model_json({
+                "model": model,
+                "messages": [{"role": "user", "content": images + [{"type": "text", "text": instruction}]}],
+                "response_format": {"type": "json_object"},
+                "temperature": 0.0,
+                "enable_thinking": False,
+                "vl_high_resolution_images": True,
+                "max_tokens": max_tokens,
+            }, f"分阶段结构图谱生成失败（{stage}）", max_attempts=1,
+               timeout_seconds=timeout_seconds, decoder=lambda text: json.loads(text))
+        except Exception as exc:
+            row.update({
+                "status": "failed", "elapsed_seconds": round(time.time() - started, 3),
+                "error": f"{type(exc).__name__}: {exc}",
+            })
+            raise PartitionedGraphError(f"{stage}: {exc}", audit) from exc
+        row.update({"status": "received", "elapsed_seconds": round(time.time() - started, 3)})
+        if not isinstance(result, dict):
+            row["status"] = "rejected"
+            row["error"] = "response_not_object"
+            raise PartitionedGraphError(f"{stage}: 模型返回必须是JSON对象", audit)
+        return result, row
+
+    def fail(stage: str, row: dict, detail: str):
+        row.update({"status": "rejected", "error": detail})
+        raise PartitionedGraphError(f"{stage}: {detail}", audit)
+
+    core_prompt = (
+        "你正在执行家具CAD结构图谱的第1/4段：只建立证据、参数、整体包围、总成和零件定义。"
+        "必须重新查看全部原图；不得输出instances、connections、assembly_connections，也不得输出成本。"
+        "每个part的observed_quantity必须是正整数，表示原图支持的物理件数量；不确定结构写入uncertainties，"
+        "不能用常见家具经验补件。固定schema_version=furniture-assembly-graph/2、"
+        "placement_solver=anchored_axes_v1、orientation_solver=role_axis_templates_v1、"
+        "contact_derivation=aabb_within_assembly_v1。零件size_mm直接使用产品X/Y/Z坐标；"
+        "每个独立运动门或抽屉建立独立assembly，assembly边界必须紧贴其结构实体。"
+        "只返回一份图谱核心JSON，字段为schema_version、placement_solver、orientation_solver、"
+        "contact_derivation、observations、parameters、envelope_mm、可选reference_bounds_mm、assemblies、"
+        "parts、uncertainties。不得引用公司BOM。\n可选五金模板合同："
+        + json.dumps(hardware_contract, ensure_ascii=False, separators=(",", ":"))
+        + evidence_context
+    )
+    core, core_audit = request("core", core_prompt, max_tokens=6000, timeout_seconds=180)
+    required = {
+        "schema_version", "placement_solver", "orientation_solver", "contact_derivation",
+        "observations", "parameters", "envelope_mm", "assemblies", "parts", "uncertainties",
+    }
+    missing = sorted(required - set(core))
+    if missing:
+        fail("core", core_audit, f"缺少核心字段: {','.join(missing)}")
+    if core.get("schema_version") != SCHEMA_VERSION:
+        fail("core", core_audit, "schema_version不匹配")
+    for forbidden in ("instances", "connections", "assembly_connections"):
+        if core.get(forbidden) not in (None, []):
+            fail("core", core_audit, f"第1段越权输出{forbidden}")
+    if not isinstance(core.get("assemblies"), list) or not isinstance(core.get("parts"), list):
+        fail("core", core_audit, "assemblies和parts必须是数组")
+    assembly_ids = [str(row.get("id") or "") for row in core["assemblies"] if isinstance(row, dict)]
+    part_ids = [str(row.get("id") or "") for row in core["parts"] if isinstance(row, dict)]
+    if not assembly_ids or "" in assembly_ids or len(set(assembly_ids)) != len(assembly_ids):
+        fail("core", core_audit, "总成ID缺失或重复")
+    if not part_ids or "" in part_ids or len(set(part_ids)) != len(part_ids):
+        fail("core", core_audit, "零件ID缺失或重复")
+    for part in core["parts"]:
+        quantity = part.get("observed_quantity") if isinstance(part, dict) else None
+        if isinstance(quantity, bool) or not isinstance(quantity, int) or quantity < 1:
+            fail("core", core_audit, f"零件{part.get('id') if isinstance(part, dict) else '?'}缺少正整数observed_quantity")
+    core_audit.update({
+        "status": "accepted", "assembly_count": len(assembly_ids), "part_count": len(part_ids),
+    })
+
+    part_by_id = {row["id"]: row for row in core["parts"]}
+    structural_parts = {
+        part_id: row for part_id, row in part_by_id.items()
+        if row.get("functional_role") != "hardware"
+    }
+    hardware_parts = {
+        part_id: row for part_id, row in part_by_id.items()
+        if row.get("functional_role") == "hardware"
+    }
+    structural_prompt = (
+        "你正在执行家具CAD结构图谱的第2/4段：根据全部原图和已锁定核心，只输出每个非五金物理实例。"
+        "返回{\"instances\":[...]}，数量必须逐part精确等于observed_quantity。每个实例必须有唯一id、"
+        "part_id、assembly_id和x/y/z三轴placement。每轴只允许absolute(value_mm)，或"
+        "align(self_anchor,target,target_anchor,offset_mm)。为消除跨总成歧义，本段align的target"
+        "只允许envelope、字面量assembly或该实例自己的assembly_id；禁止引用其他总成或其他实例。"
+        "门板/抽屉部件必须在自己的紧边界assembly内定位；不得输出五金、mountings、connections或成本。"
+        "不得引用公司BOM。\n已锁定核心："
+        + json.dumps(core, ensure_ascii=False, separators=(",", ":")) + evidence_context
+    )
+    structural_response, structural_audit = request(
+        "structural_instances", structural_prompt, max_tokens=5000, timeout_seconds=180,
+    )
+    structural_instances = structural_response.get("instances")
+    if not isinstance(structural_instances, list):
+        fail("structural_instances", structural_audit, "缺少instances数组")
+    seen_instance_ids = set()
+    structural_counts = {}
+    for item in structural_instances:
+        if not isinstance(item, dict):
+            fail("structural_instances", structural_audit, "结构实例必须是对象")
+        instance_id = str(item.get("id") or "")
+        part_id = str(item.get("part_id") or "")
+        assembly_id = str(item.get("assembly_id") or "")
+        if not instance_id or instance_id in seen_instance_ids:
+            fail("structural_instances", structural_audit, "结构实例ID缺失或重复")
+        if part_id not in structural_parts or assembly_id not in assembly_ids:
+            fail("structural_instances", structural_audit, f"实例{instance_id}引用了非法零件或总成")
+        placement = item.get("placement")
+        if not isinstance(placement, dict) or set(placement) != {"x", "y", "z"}:
+            fail("structural_instances", structural_audit, f"实例{instance_id}必须声明三轴placement")
+        for axis in "xyz":
+            rule = placement[axis]
+            if not isinstance(rule, dict) or rule.get("mode") not in {"absolute", "align"}:
+                fail("structural_instances", structural_audit, f"实例{instance_id}的{axis}轴定位无效")
+            if rule.get("mode") == "align" and rule.get("target") not in {"envelope", "assembly", assembly_id}:
+                fail("structural_instances", structural_audit, f"实例{instance_id}的{axis}轴发生跨总成或实例引用")
+        seen_instance_ids.add(instance_id)
+        structural_counts[part_id] = structural_counts.get(part_id, 0) + 1
+    for part_id, part in structural_parts.items():
+        if structural_counts.get(part_id, 0) != part["observed_quantity"]:
+            fail(
+                "structural_instances", structural_audit,
+                f"零件{part_id}实例数{structural_counts.get(part_id, 0)}不等于证据数量{part['observed_quantity']}",
+            )
+    structural_audit.update({"status": "accepted", "instance_count": len(structural_instances)})
+
+    hardware_instances = []
+    if hardware_parts:
+        hardware_prompt = (
+            "你正在执行家具CAD结构图谱的第3/4段：只生成五金实例和安装关系。必须回看全部原图。"
+            "返回{\"instances\":[...]}，每个hardware part的实例数必须等于observed_quantity。"
+            "placement仅为1×1×1占位，x/y/z都必须使用absolute；真实几何完全由模板mountings决定。"
+            "每个mounting的host_instance_id只能引用给出的结构实例ID，slot、kind、face、"
+            "embed_depth_mm、mounting_center_mm和evidence_ids必须齐全并服从模板合同。"
+            "没有图像证据不得增加五金。不得输出结构实例、总成连接或成本，不得引用公司BOM。\n核心："
+            + json.dumps(core, ensure_ascii=False, separators=(",", ":"))
+            + "\n合法结构实例ID：" + json.dumps(sorted(seen_instance_ids), ensure_ascii=False)
+            + "\n五金模板合同：" + json.dumps(hardware_contract, ensure_ascii=False, separators=(",", ":"))
+            + evidence_context
+        )
+        hardware_response, hardware_audit = request(
+            "hardware_instances", hardware_prompt, max_tokens=3500, timeout_seconds=150,
+        )
+        hardware_instances = hardware_response.get("instances")
+        if not isinstance(hardware_instances, list):
+            fail("hardware_instances", hardware_audit, "缺少instances数组")
+        hardware_counts = {}
+        for item in hardware_instances:
+            if not isinstance(item, dict):
+                fail("hardware_instances", hardware_audit, "五金实例必须是对象")
+            instance_id = str(item.get("id") or "")
+            part_id = str(item.get("part_id") or "")
+            assembly_id = str(item.get("assembly_id") or "")
+            if not instance_id or instance_id in seen_instance_ids:
+                fail("hardware_instances", hardware_audit, "五金实例ID缺失或与结构实例重复")
+            if part_id not in hardware_parts or assembly_id not in assembly_ids:
+                fail("hardware_instances", hardware_audit, f"五金实例{instance_id}引用了非法零件或总成")
+            placement = item.get("placement")
+            if not isinstance(placement, dict) or set(placement) != {"x", "y", "z"} or any(
+                not isinstance(placement[axis], dict) or placement[axis].get("mode") != "absolute"
+                for axis in "xyz"
+            ):
+                fail("hardware_instances", hardware_audit, f"五金实例{instance_id}占位定位必须为三轴absolute")
+            mountings = item.get("mountings")
+            if not isinstance(mountings, list) or not mountings:
+                fail("hardware_instances", hardware_audit, f"五金实例{instance_id}缺少模板安装位")
+            if any(str(mount.get("host_instance_id") or "") not in seen_instance_ids for mount in mountings if isinstance(mount, dict)):
+                fail("hardware_instances", hardware_audit, f"五金实例{instance_id}安装宿主不在结构实例中")
+            seen_instance_ids.add(instance_id)
+            hardware_counts[part_id] = hardware_counts.get(part_id, 0) + 1
+        for part_id, part in hardware_parts.items():
+            if hardware_counts.get(part_id, 0) != part["observed_quantity"]:
+                fail(
+                    "hardware_instances", hardware_audit,
+                    f"五金{part_id}实例数{hardware_counts.get(part_id, 0)}不等于证据数量{part['observed_quantity']}",
+                )
+        hardware_audit.update({"status": "accepted", "instance_count": len(hardware_instances)})
+    else:
+        audit.append({
+            "stage": "hardware_instances", "status": "skipped_no_hardware_parts",
+            "source_images_included": False, "company_bom_used_in_inference": False,
+        })
+
+    assembly_connections = []
+    if len(assembly_ids) > 1:
+        relation_context = [{
+            "id": row["id"], "role": row.get("role"), "bounds_mm": row.get("bounds_mm"),
+            "member_instance_ids": sorted(
+                item["id"] for item in structural_instances + hardware_instances
+                if item.get("assembly_id") == row["id"]
+            ),
+        } for row in core["assemblies"]]
+        relation_prompt = (
+            "你正在执行家具CAD结构图谱的第4/4段：只生成总成之间的真实装配/运动关系。"
+            "必须回看全部原图，返回{\"assembly_connections\":[...]}。a和b只能引用给出的assembly ID；"
+            "每个door或drawer运动总成必须且只能出现一次并放在a端，按图像证据选择hinged或sliding；"
+            "hinged给axis与angle_limits_deg，sliding给axis与travel_limits_mm。固定模块仅在真实接触时用fixed。"
+            "不能引用实例ID，不能输出零件、定位或成本，不得引用公司BOM。\n总成摘要："
+            + json.dumps(relation_context, ensure_ascii=False, separators=(",", ":")) + evidence_context
+        )
+        relation_response, relation_audit = request(
+            "assembly_connections", relation_prompt, max_tokens=2500, timeout_seconds=120,
+        )
+        assembly_connections = relation_response.get("assembly_connections")
+        if not isinstance(assembly_connections, list):
+            fail("assembly_connections", relation_audit, "缺少assembly_connections数组")
+        connection_ids = set()
+        moving_counts = {row["id"]: 0 for row in core["assemblies"] if row.get("role") in {"door", "drawer"}}
+        for item in assembly_connections:
+            if not isinstance(item, dict):
+                fail("assembly_connections", relation_audit, "总成连接必须是对象")
+            connection_id = str(item.get("id") or "")
+            if not connection_id or connection_id in connection_ids:
+                fail("assembly_connections", relation_audit, "总成连接ID缺失或重复")
+            a_id, b_id = str(item.get("a") or ""), str(item.get("b") or "")
+            if a_id not in assembly_ids or b_id not in assembly_ids or a_id == b_id:
+                fail("assembly_connections", relation_audit, f"总成连接{connection_id}端点非法")
+            if a_id in moving_counts:
+                moving_counts[a_id] += 1
+                if item.get("kind") not in {"hinged", "sliding"}:
+                    fail("assembly_connections", relation_audit, f"运动总成{a_id}缺少hinged/sliding关系")
+            elif b_id in moving_counts:
+                fail("assembly_connections", relation_audit, f"运动总成{b_id}必须位于关系a端")
+            connection_ids.add(connection_id)
+        invalid_moving = {key: count for key, count in moving_counts.items() if count != 1}
+        if invalid_moving:
+            fail("assembly_connections", relation_audit, f"运动总成关系数量不等于1: {invalid_moving}")
+        relation_audit.update({"status": "accepted", "connection_count": len(assembly_connections)})
+    else:
+        audit.append({
+            "stage": "assembly_connections", "status": "skipped_single_assembly",
+            "source_images_included": False, "company_bom_used_in_inference": False,
+        })
+
+    graph = copy.deepcopy(core)
+    graph["instances"] = copy.deepcopy(structural_instances + hardware_instances)
+    graph["connections"] = []
+    graph["assembly_connections"] = copy.deepcopy(assembly_connections)
+    return graph, audit
+
+
 def render_multiview(ai3d, build_dir: pathlib.Path, step_path: str) -> list[dict]:
     """Render orthographic evidence so the reviewer is not trapped in one ISO view."""
     python = str(ai3d.runtime_dir / "venv/bin/python")
@@ -1305,17 +1584,27 @@ def main():
                 "blocking": report["source_qualification"]["blocking"],
             }, ensure_ascii=False), flush=True)
             continue
+        compact_source_structure = {
+            key: source_structure.get(key)
+            for key in (
+                "drawers", "doors", "back_panel", "parts", "genuinely_invisible",
+                "source_completeness_issues",
+            )
+            if source_structure.get(key) not in (None, [], {})
+        }
+        compact_source_structure["accepted_cross_view_facts"] = (
+            (source_structure.get("view_adjudication") or {}).get("accepted") or {}
+        )
         evidence_context = (
             "\n以下是同一批原图经过两个独立AI专项读取后的候选证据。它们不是公司BOM，也不是最终答案；"
             "你必须回看原图。dimension_ledger中存在的X/Y/Z是已通过范围、方向和置信度门的整体尺寸，必须精确采用；"
             "structure.view_adjudication.accepted是跨视图成立的结构约束；conflicts和machine_followups应由本轮图像复核解决，"
             "不能直接转嫁给用户。\n" + json.dumps({
-                "structure": source_structure,
-                "dimension_read": source_dimensions,
+                "structure": compact_source_structure,
                 "dimension_ledger": dimension_ledger,
                 "topology_contract": topology_contract,
                 "verified_scope_facts": scope_facts,
-            }, ensure_ascii=False)
+            }, ensure_ascii=False, separators=(",", ":"))
         )
         graph, feedback = {}, []
         accepted_compile_issue_count = None
@@ -1574,15 +1863,28 @@ def main():
                     + evidence_context
                     + "\n编译器及原图视觉复核的结构化反馈：" + json.dumps(feedback, ensure_ascii=False)
                 )
-            phase = "ai_graph_from_original_images" if not graph else "ai_local_graph_repair_from_feedback"
+            phase = "ai_partitioned_graph_from_original_images" if not graph else "ai_local_graph_repair_from_feedback"
             print(json.dumps({"case": case, "round": row["round"], "phase": phase}), flush=True)
             try:
-                response = engine._request_model_json({
-                    "model": args.model, "messages": [{"role": "user", "content": images + [{"type": "text", "text": prompt}]}],
-                    "response_format": {"type": "json_object"}, "temperature": 0.0,
-                    "enable_thinking": True, "thinking_budget": 2048, "vl_high_resolution_images": True, "max_tokens": 14000,
-                }, "原图到参数化结构图谱失败", max_attempts=1, timeout_seconds=180,
-                   decoder=lambda text: json.loads(text))
+                if graph:
+                    request_body = {
+                        "model": args.model,
+                        "messages": [{"role": "user", "content": images + [{"type": "text", "text": prompt}]}],
+                        "response_format": {"type": "json_object"}, "temperature": 0.0,
+                        "enable_thinking": True, "thinking_budget": 2048,
+                        "vl_high_resolution_images": True, "max_tokens": 6000,
+                    }
+                    response = engine._request_model_json(
+                        request_body, "原图到参数化结构图谱失败", max_attempts=1,
+                        timeout_seconds=180, decoder=lambda text: json.loads(text),
+                    )
+                else:
+                    response, partition_audit = generate_partitioned_initial_graph(
+                        engine=engine, model=args.model, images=images,
+                        evidence_context=evidence_context,
+                        hardware_contract=supported_hardware_contract(),
+                    )
+                    row["initial_graph_partition_audit"] = partition_audit
                 if graph:
                     row["patch_response"] = response
                     previous_graph = graph
@@ -1742,6 +2044,8 @@ def main():
                         report["status"] = "graph_not_passed"
             except Exception as exc:
                 row["error"] = f"{type(exc).__name__}: {exc}"
+                if isinstance(exc, PartitionedGraphError):
+                    row["initial_graph_partition_audit"] = exc.audit
                 dependency_feedback = terminal_dependency_feedback(exc)
                 if dependency_feedback:
                     row["terminal_dependency_error"] = dependency_feedback

server_patch/ai_3d_model/ai-solid-wood-cost/solid_wood_cost_engine.py

@@ -449,7 +449,7 @@ class SolidWoodCostEngine:
                 "/opt/dalongxia-chat-api/ai-3d-model/ai_3d_model_engine.py",
             )
         )
-        self.model = os.environ.get("DALONGXIA_SOLID_WOOD_VISION_MODEL", "qwen-vl-plus")
+        self.model = os.environ.get("DALONGXIA_SOLID_WOOD_VISION_MODEL", "qwen3-vl-plus")
         self.api_base = os.environ.get(
             "DALONGXIA_AI_3D_API_BASE",
             "https://dashscope.aliyuncs.com/compatible-mode/v1",

server_patch/ai_3d_model/ai-solid-wood-cost/structure_graph.py

@@ -1100,16 +1100,21 @@ def _solve_anchored_axis_placements(
     resolver: ParameterResolver,
     envelope_mm: list[float],
     assembly_bounds: dict | None = None,
-) -> tuple[dict, list[dict]]:
+) -> tuple[dict, list[dict], list[dict]]:
     """Resolve declared design intent into coordinates.
 
     The model names an anchor relationship (for example, the right side's
     maximum X is aligned to the product envelope's maximum X).  It never gets
     to guess the resulting coordinate.  References form a dependency graph;
     cycles and missing axes are rejected instead of being numerically nudged.
+
+    Placement failures are isolated to the affected instance dependency tree.
+    A single invalid member must not erase already valid geometry and trigger a
+    misleading cascade of empty-assembly, quantity and topology errors.
     """
     resolved = {}
     audit = []
+    failures = {}
     visiting = set()
     anchors = {"min": 0.0, "center": 0.5, "max": 1.0}
 
@@ -1119,6 +1124,8 @@ def _solve_anchored_axis_placements(
     def solve(instance_id):
         if instance_id in resolved:
             return resolved[instance_id]
+        if instance_id in failures:
+            raise GraphError(failures[instance_id])
         if instance_id in visiting:
             raise GraphError(f"实例定位依赖成环: {instance_id}")
         if instance_id not in instance_specs:
@@ -1130,6 +1137,7 @@ def _solve_anchored_axis_placements(
             if not isinstance(placement, dict) or set(placement) != {"x", "y", "z"}:
                 raise GraphError(f"实例 {instance_id} 必须分别声明X/Y/Z三轴定位约束")
             position = []
+            instance_audit = []
             for axis, axis_name in enumerate("xyz"):
                 rule = placement[axis_name]
                 if not isinstance(rule, dict):
@@ -1176,20 +1184,35 @@ def _solve_anchored_axis_placements(
                 if not math.isfinite(coordinate):
                     raise GraphError(f"实例 {instance_id} 的{axis_name.upper()}轴定位不是有限数")
                 position.append(coordinate)
-                audit.append({
+                instance_audit.append({
                     "instance_id": instance_id, "axis": axis_name,
                     "coordinate_mm": coordinate, "mode": mode, "target": target,
                     "target_kind": target_kind,
                     "source": "declared_anchor_constraint",
                 })
             resolved[instance_id] = {**spec, "position_mm": position}
+            audit.extend(instance_audit)
             return resolved[instance_id]
         finally:
             visiting.remove(instance_id)
 
     for key in instance_specs:
-        solve(key)
-    return resolved, audit
+        if key in resolved:
+            continue
+        try:
+            solve(key)
+        except (GraphError, KeyError, TypeError) as exc:
+            failures[key] = str(exc)
+    errors = [
+        {
+            "instance_id": instance_id,
+            "part_id": instance_specs.get(instance_id, {}).get("part_id"),
+            "assembly_id": instance_specs.get(instance_id, {}).get("assembly_id"),
+            "detail": detail,
+        }
+        for instance_id, detail in failures.items()
+    ]
+    return resolved, audit, errors
 
 
 def compile_graph(
@@ -1414,14 +1437,35 @@ def compile_graph(
         except (GraphError, KeyError, TypeError) as exc:
             issue("instance_geometry", f"/instances/{index}", str(exc))
     placement_audit = []
+    placement_errors = []
     if graph.get("placement_solver") == "anchored_axes_v1" and envelope:
         try:
-            instances, placement_audit = _solve_anchored_axis_placements(
+            instances, placement_audit, placement_errors = _solve_anchored_axis_placements(
                 instance_specs, resolver=resolver, envelope_mm=envelope,
                 assembly_bounds=assembly_bounds,
             )
+            for row in placement_errors:
+                instance_id = row["instance_id"]
+                issue(
+                    "placement_constraint", f"/instances/{instance_id}/placement",
+                    row["detail"], instance_id=instance_id,
+                    part_id=row.get("part_id"), assembly_id=row.get("assembly_id"),
+                )
         except (GraphError, KeyError, TypeError) as exc:
             issue("placement_constraint", "/instances", str(exc))
+    failed_instance_ids = {row["instance_id"] for row in placement_errors}
+    failed_part_ids = {
+        instance_specs[instance_id].get("part_id") for instance_id in failed_instance_ids
+        if instance_id in instance_specs
+    }
+    failed_assembly_ids = {
+        instance_specs[instance_id].get("assembly_id") for instance_id in failed_instance_ids
+        if instance_id in instance_specs
+    }
+    failed_roles = {
+        instance_specs[instance_id].get("functional_role") for instance_id in failed_instance_ids
+        if instance_id in instance_specs
+    }
     mounting_audit, mounting_issues, hardware_geometry = _validate_hardware_mountings(
         instances, observations=observations, resolver=resolver,
     )
@@ -1494,6 +1538,11 @@ def compile_graph(
                 )
         used = Counter(inst.get("assembly_id") for inst in instances.values())
         for key in assemblies:
+            if key in failed_assembly_ids:
+                # The localized placement issue is the primary repair target.
+                # Tightness and emptiness cannot be judged until all declared
+                # members of this assembly have valid coordinates.
+                continue
             if not used[key]:
                 issue("empty_assembly", f"/assemblies/{key}", "空总成不能作为装配关系的虚拟占位")
                 continue
@@ -1532,6 +1581,8 @@ def compile_graph(
         issues.extend(solver_errors)
     quantities = Counter(inst["part_id"] for inst in instances.values())
     for key, part in parts.items():
+        if key in failed_part_ids:
+            continue
         if not quantities[key]:
             issue("uninstantiated_part", f"/parts/{key}", "零件没有物理实例，不能计入BOM")
         expected = part.get("observed_quantity")
@@ -1554,6 +1605,8 @@ def compile_graph(
                 issue("source_topology_contract", "/topology_contract/role_counts", "原图拓扑契约数量格式无效")
                 continue
             actual = role_quantities.get(role, 0)
+            if role in failed_roles:
+                continue
             if actual != expected:
                 issue(
                     "source_topology_count_conflict",
@@ -1578,6 +1631,8 @@ def compile_graph(
         seen_connections.add(key)
         a, b = instances.get(edge.get("a")), instances.get(edge.get("b"))
         if not a or not b or a["id"] == b["id"]:
+            if edge.get("a") in failed_instance_ids or edge.get("b") in failed_instance_ids:
+                continue
             issue("connection_target", path, "连接两端必须是不同且存在的物理实例")
             continue
         if edge.get("kind") not in {"fixed", "sliding", "hinged", "support"}:

server_patch/ai_3d_model/ai-solid-wood-cost/test_manufacturing_oracle.py

@@ -10,12 +10,14 @@ from unittest import mock
 
 from manufacturing_oracle import plan_manufacturing_features
 from probe_structure_graph import (
+    PartitionedGraphError,
     authoritative_interference_feedback, bind_scope_facts,
     build_and_authoritatively_validate, 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,
     enrich_hardware_feedback_with_verified_placements,
+    generate_partitioned_initial_graph,
     normalize_focused_source_adjudication, normalize_scope_facts,
     qualified_occupied_scope_adjudication,
     normalize_visual_review,
@@ -75,6 +77,124 @@ def compile_case(graph):
 
 
 class ManufacturingOracleTest(unittest.TestCase):
+    @staticmethod
+    def _partitioned_graph_responses(*, invalid_door_target=False):
+        absolute = {
+            "x": {"mode": "absolute", "value_mm": 0},
+            "y": {"mode": "absolute", "value_mm": 0},
+            "z": {"mode": "absolute", "value_mm": 0},
+        }
+        door_placement = copy.deepcopy(absolute)
+        if invalid_door_target:
+            door_placement["x"] = {
+                "mode": "align", "self_anchor": "min", "target": "carcass",
+                "target_anchor": "min", "offset_mm": 0,
+            }
+        core = {
+            "schema_version": SCHEMA_VERSION,
+            "placement_solver": "anchored_axes_v1",
+            "orientation_solver": "role_axis_templates_v1",
+            "contact_derivation": "aabb_within_assembly_v1",
+            "observations": [{
+                "id": "view", "source_type": "image", "image_id": "image1",
+                "confidence": 0.95,
+            }],
+            "parameters": {}, "envelope_mm": [800, 400, 700],
+            "assemblies": [
+                {"id": "carcass", "role": "carcass", "bounds_mm": {
+                    "x": [0, 800], "y": [0, 400], "z": [0, 700],
+                }, "evidence_ids": ["view"]},
+                {"id": "door-1", "role": "door", "bounds_mm": {
+                    "x": [18, 782], "y": [0, 18], "z": [0, 700],
+                }, "evidence_ids": ["view"]},
+            ],
+            "parts": [
+                {"id": "side", "name": "侧板", "shape": "box", "functional_role": "side",
+                 "thickness_axis": "x", "size_mm": [18, 400, 700], "observed_quantity": 2},
+                {"id": "door", "name": "门板", "shape": "box", "functional_role": "door",
+                 "thickness_axis": "y", "size_mm": [764, 18, 700], "observed_quantity": 1},
+                {"id": "knob", "name": "旋钮", "shape": "box", "functional_role": "hardware",
+                 "thickness_axis": None, "hardware_family": "knob",
+                 "hardware_template_id": "knob_m4_surface_v1", "size_mm": [1, 1, 1],
+                 "observed_quantity": 1},
+            ],
+            "uncertainties": [],
+        }
+        structural = {"instances": [
+            {"id": "side-left", "part_id": "side", "assembly_id": "carcass",
+             "placement": copy.deepcopy(absolute)},
+            {"id": "side-right", "part_id": "side", "assembly_id": "carcass",
+             "placement": copy.deepcopy(absolute)},
+            {"id": "door-panel", "part_id": "door", "assembly_id": "door-1",
+             "placement": door_placement},
+        ]}
+        hardware = {"instances": [{
+            "id": "knob-1", "part_id": "knob", "assembly_id": "door-1",
+            "placement": copy.deepcopy(absolute),
+            "mountings": [{
+                "slot": "body", "host_instance_id": "door-panel", "kind": "surface",
+                "face": "y-", "embed_depth_mm": 0,
+                "mounting_center_mm": [650, 0, 350], "evidence_ids": ["view"],
+            }],
+        }]}
+        relations = {"assembly_connections": [{
+            "id": "door-motion", "kind": "hinged", "a": "door-1", "b": "carcass",
+            "axis": "z", "angle_limits_deg": [0, 110], "evidence_ids": ["view"],
+        }]}
+        return [core, structural, hardware, relations]
+
+    def test_initial_graph_generation_is_partitioned_and_reference_safe(self):
+        class FakeEngine:
+            def __init__(self, responses):
+                self.responses = list(responses)
+                self.calls = []
+
+            def _request_model_json(self, body, *_args, **_kwargs):
+                self.calls.append(body)
+                return copy.deepcopy(self.responses.pop(0))
+
+        engine = FakeEngine(self._partitioned_graph_responses())
+        source_images = [{"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}}]
+        graph, audit = generate_partitioned_initial_graph(
+            engine=engine, model="test-vl", images=source_images,
+            evidence_context="\nINDEPENDENT_SOURCE_EVIDENCE", hardware_contract={},
+        )
+
+        self.assertEqual(4, len(engine.calls))
+        self.assertEqual(
+            ["core", "structural_instances", "hardware_instances", "assembly_connections"],
+            [row["stage"] for row in audit],
+        )
+        self.assertTrue(all(row["status"] == "accepted" for row in audit))
+        self.assertTrue(all(row["company_bom_used_in_inference"] is False for row in audit))
+        self.assertEqual(4, len(graph["instances"]))
+        self.assertEqual([], graph["connections"])
+        self.assertEqual("door-1", graph["assembly_connections"][0]["a"])
+        for call in engine.calls:
+            content = call["messages"][0]["content"]
+            self.assertEqual(source_images[0], content[0])
+            self.assertNotIn("COMPANY_BOM_SENTINEL", json.dumps(content, ensure_ascii=False))
+            self.assertFalse(call["enable_thinking"])
+
+    def test_partition_gate_rejects_cross_assembly_target_before_merge(self):
+        class FakeEngine:
+            def __init__(self, responses):
+                self.responses = list(responses)
+                self.calls = []
+
+            def _request_model_json(self, body, *_args, **_kwargs):
+                self.calls.append(body)
+                return copy.deepcopy(self.responses.pop(0))
+
+        engine = FakeEngine(self._partitioned_graph_responses(invalid_door_target=True))
+        with self.assertRaisesRegex(PartitionedGraphError, "跨总成或实例引用") as raised:
+            generate_partitioned_initial_graph(
+                engine=engine, model="test-vl", images=[],
+                evidence_context="\nINDEPENDENT_SOURCE_EVIDENCE", hardware_contract={},
+            )
+        self.assertEqual(2, len(engine.calls))
+        self.assertEqual("rejected", raised.exception.audit[-1]["status"])
+
     def test_missing_model_credentials_are_terminal_dependency_not_kernel_feedback(self):
         feedback = terminal_dependency_feedback(RuntimeError("家具读图模型尚未配置密钥"))
         self.assertEqual("model_credentials_missing", feedback["code"])
@@ -88,6 +208,12 @@ class ManufacturingOracleTest(unittest.TestCase):
         self.assertEqual("model_output_truncated", truncated["code"])
         self.assertEqual("model_gateway_capacity", truncated["next_owner"])
         self.assertFalse(truncated["retryable"])
+        timed_out = terminal_dependency_feedback(RuntimeError(
+            "原图到参数化结构图谱失败：The read operation timed out"
+        ))
+        self.assertEqual("model_request_timeout", timed_out["code"])
+        self.assertEqual("request_partitioning_or_model_routing", timed_out["next_owner"])
+        self.assertFalse(timed_out["retryable"])
         self.assertIsNone(terminal_dependency_feedback(RuntimeError("ordinary compile error")))
 
     def test_visual_review_drops_self_declared_non_issues_and_unproven_proportion_guess(self):

server_patch/ai_3d_model/ai-solid-wood-cost/test_structure_graph.py

@@ -670,6 +670,26 @@ class StructureGraphTest(unittest.TestCase):
         invalid["instances"][0]["placement"]["x"]["target"] = "other-assembly"
         self.assertIn("placement_constraint", {row["code"] for row in compile_case(invalid)["issues"]})
 
+    def test_invalid_anchor_is_localized_and_preserves_unrelated_geometry(self):
+        graph = anchored_frame()
+        graph["instances"][0]["placement"]["x"]["target"] = "other-assembly"
+
+        result = compile_case(graph)
+
+        self.assertFalse(result["ok"])
+        self.assertEqual(["i2"], [row["id"] for row in result["instances"]])
+        self.assertEqual(3, len(result["placement_audit"]))
+        placement_issues = [
+            row for row in result["issues"] if row["code"] == "placement_constraint"
+        ]
+        self.assertEqual(
+            {"/instances/i1/placement", "/instances/i3/placement"},
+            {row["path"] for row in placement_issues},
+        )
+        self.assertNotIn("/instances", {row["path"] for row in placement_issues})
+        cascade_codes = {"empty_assembly", "uninstantiated_part", "quantity_conflict"}
+        self.assertFalse(cascade_codes & {row["code"] for row in result["issues"]})
+
     def test_anchor_constraints_follow_parameter_changes(self):
         graph = anchored_frame()
         first = compile_case(graph)