step_editor/scdm_result_validator.py

from __future__ import annotations

import math
from collections.abc import Callable, Iterable, Mapping, Sequence
from pathlib import Path

from .relation_formulas import rewrite_relation_formula_ids


def validate_scdm_edit_result(
    edit_result: Mapping[str, object],
    *,
    before_signature: Mapping[str, object] | None = None,
    before_cache: Mapping[str, object] | None = None,
    after_cache: Mapping[str, object] | None = None,
    capability_key: str = "",
    expected_target: object = None,
    edited_object_id: str = "",
    tolerance: float = 1.0e-6,
    brep_validator: Callable[[Path], Mapping[str, object]] | None = None,
) -> dict[str, object]:
    if edit_result.get("ok") is not True:
        return {
            "ok": False,
            "reason": str(edit_result.get("reason") or "edit-failed"),
            "message": str(edit_result.get("message") or "SCDM edit did not succeed."),
            "editResult": dict(edit_result),
        }

    output_step = Path(str(edit_result.get("output_step") or edit_result.get("outputStep") or "")).expanduser()
    if not output_step.is_file():
        return {
            "ok": False,
            "reason": "missing-output-step",
            "message": f"SCDM result STEP does not exist: {output_step}",
            "editResult": dict(edit_result),
        }

    if brep_validator is not None:
        brep = dict(brep_validator(output_step))
        if brep.get("ok") is not True:
            return {
                "ok": False,
                "reason": str(brep.get("reason") or "brep-invalid"),
                "message": str(brep.get("message") or "OCCT rejected the result STEP."),
                "brep": brep,
                "editResult": dict(edit_result),
            }
    else:
        brep = {"ok": None, "reason": "not-run", "message": "B-Rep validation callback was not provided."}

    summary_check: dict[str, object] = {"ok": None, "reason": "not-run", "message": "SCDM summary check needs the old and new caches."}
    topology_check: dict[str, object] = {"ok": None, "reason": "not-run", "message": "Object drift check needs the old and new SCDM caches."}
    context_failure: dict[str, object] | None = None
    if before_cache and after_cache:
        # B-Rep 有效只说明 STEP 能读回来；还要比较 SCDM cache，防止一次局部编辑
        # 顺手让其它可识别对象消失、漂移或变得歧义。
        summary_check = check_scdm_summary_delta(before_cache, after_cache, capability_key=capability_key)
        if summary_check.get("ok") is False:
            context_failure = {
                "ok": False,
                "reason": str(summary_check.get("reason") or "summary-drift"),
                "message": str(summary_check.get("message") or "SCDM result changed the model summary too much."),
                "summaryCheck": summary_check,
                "brep": brep,
                "editResult": dict(edit_result),
            }
        topology_check = check_scdm_unedited_objects(
            before_cache,
            after_cache,
            edited_object_id=edited_object_id,
            edited_signature=before_signature,
            capability_key=capability_key,
        )
        if topology_check.get("ok") is not True:
            if context_failure is None:
                context_failure = {
                    "ok": False,
                    "reason": str(topology_check.get("reason") or "unexpected-object-drift"),
                    "message": str(topology_check.get("message") or "SCDM result changed unrelated recognized objects."),
                    "topologyCheck": topology_check,
                    "brep": brep,
                    "editResult": dict(edit_result),
                }

    if _is_removal_capability(capability_key) and (not before_signature or not after_cache):
        return {
            "ok": False,
            "reason": "removal-check-unavailable",
            "message": "Removed feature cannot be verified without the old feature signature and the new SCDM cache.",
            "summaryCheck": summary_check,
            "topologyCheck": topology_check,
            "brep": brep,
            "editResult": dict(edit_result),
        }

    if capability_key == "pattern.segment_spacing":
        check = _check_pattern_segment_spacing_edit_result(
            edit_result,
            expected_target,
            before_signature=before_signature,
            tolerance=tolerance,
        )
        if check.get("ok") is not True:
            return {
                "ok": False,
                "reason": str(check.get("reason") or "target-check-failed"),
                "message": str(check.get("message") or "SCDM result did not reach the target segment spacing."),
                "targetCheck": check,
                "summaryCheck": summary_check,
                "topologyCheck": topology_check,
                "brep": brep,
                "editResult": dict(edit_result),
            }
        warning = _context_guard_warning_if_target_verified(
            context_failure,
            capability_key=capability_key,
            target_check=check,
            summary_check=summary_check,
            brep=brep,
        )
        if context_failure is not None and warning is None:
            return context_failure
        return {
            "ok": True,
            "reason": "ok",
            "message": "SCDM edit result passed the available validation checks.",
            "output_step": str(output_step),
            "matchedObject": None,
            "targetCheck": check,
            "removalCheck": {"ok": None, "reason": "not-run", "message": "Removal check is not needed for this capability."},
            "summaryCheck": summary_check,
            "topologyCheck": topology_check,
            "brep": brep,
            "editResult": dict(edit_result),
            "validationWarnings": [warning] if warning else [],
        }

    matched: dict[str, object] | None = None
    removal_check: dict[str, object] = {"ok": None, "reason": "not-run", "message": "Removal check is not needed for this capability."}
    if before_signature and after_cache:
        # 修改后的拓扑 ID 可能变化，不能按旧 Face ID 验证；这里用几何签名在新 cache
        # 里找唯一对象，再用它做目标值回测和关系式 ID 续接。
        match = match_scdm_object_by_signature(before_signature, after_cache, capability_key=capability_key)
        status = str(match.get("status") or "")
        if _is_removal_capability(capability_key):
            removal_check = _check_removed_object_match(match)
            if removal_check.get("ok") is not True:
                return {
                    "ok": False,
                    "reason": str(removal_check.get("reason") or "feature-still-present"),
                    "message": str(removal_check.get("message") or "Removed feature is still present in the new SCDM cache."),
                    "removalCheck": removal_check,
                    "match": match,
                    "brep": brep,
                    "editResult": dict(edit_result),
                }
            check = {"ok": True, "reason": "removed", "message": "Target feature disappeared from the new SCDM cache."}
            return {
                "ok": True,
                "reason": "ok",
                "message": "SCDM edit result passed the available validation checks.",
                "output_step": str(output_step),
                "matchedObject": None,
                "targetCheck": check,
                "removalCheck": removal_check,
                "summaryCheck": summary_check,
                "topologyCheck": topology_check,
                "brep": brep,
                "editResult": dict(edit_result),
            }
        if status != "unique":
            return {
                "ok": False,
                "reason": f"object-match-{status or 'failed'}",
                "message": str(match.get("message") or "Edited object could not be uniquely matched in the new SCDM cache."),
                "match": match,
                "brep": brep,
                "editResult": dict(edit_result),
            }
        candidate = match.get("object")
        if isinstance(candidate, Mapping):
            matched = dict(candidate)

    if expected_target is not None and matched is not None:
        check = check_scdm_target(matched, capability_key=capability_key, expected_target=expected_target, tolerance=tolerance)
        if check.get("ok") is not True:
            return {
                "ok": False,
                "reason": str(check.get("reason") or "target-check-failed"),
                "message": str(check.get("message") or "SCDM result did not reach the target value."),
                "targetCheck": check,
                "matchedObject": matched,
                "brep": brep,
                "editResult": dict(edit_result),
            }
    else:
        check = {"ok": None, "reason": "not-run", "message": "Target check needs a matched object and an expected target."}

    warning = _context_guard_warning_if_target_verified(
        context_failure,
        capability_key=capability_key,
        target_check=check,
        summary_check=summary_check,
        brep=brep,
    )
    if context_failure is not None and warning is None:
        return context_failure

    return {
        "ok": True,
        "reason": "ok",
        "message": "SCDM edit result passed the available validation checks.",
        "output_step": str(output_step),
        "matchedObject": matched,
        "targetCheck": check,
        "removalCheck": removal_check,
        "summaryCheck": summary_check,
        "topologyCheck": topology_check,
        "brep": brep,
        "editResult": dict(edit_result),
        "validationWarnings": [warning] if warning else [],
    }


def _context_guard_warning_if_target_verified(
    context_failure: Mapping[str, object] | None,
    *,
    capability_key: str,
    target_check: Mapping[str, object],
    summary_check: Mapping[str, object],
    brep: Mapping[str, object],
) -> dict[str, object] | None:
    # SCDM 重新导出 STEP 时有时会重新划分 body count，但目标对象和 B-Rep 都正确。
    # 对已能目标回测的几何能力，将这种情况降级为警告，避免把成功修改误回滚。
    if brep.get("ok") is not True:
        return None
    if target_check.get("ok") is not True:
        return None
    if str(summary_check.get("reason") or "") != "body-count-repartitioned":
        return None
    if not _is_target_verified_scdm_geometry_capability(capability_key):
        return None
    if not isinstance(context_failure, Mapping):
        return {
            "reason": "body-count-repartitioned",
            "message": str(summary_check.get("message") or "SCDM changed body count while re-exporting the STEP result."),
            "summaryCheck": dict(summary_check),
        }
    warning = {
        "reason": str(context_failure.get("reason") or "context-guard-warning"),
        "message": str(context_failure.get("message") or "SCDM cache context changed after edit."),
        "summaryCheck": dict(summary_check),
    }
    topology_check = context_failure.get("topologyCheck")
    if isinstance(topology_check, Mapping):
        warning["topologyCheck"] = dict(topology_check)
    return warning


def _is_target_verified_scdm_geometry_capability(capability_key: str) -> bool:
    return capability_key in {
        "face.offset",
        "hole.diameter",
        "hole.position",
        "slot.width",
        "slot.depth",
        "slot.position",
        "boss.height",
        "boss.diameter",
        "boss.position",
        "round.radius",
        "chamfer.distance",
        "pattern.spacing",
        "pattern.segment_spacing",
        "pattern.instance_position",
        "shell.thickness",
    }


def match_scdm_object_by_signature(
    before_signature: Mapping[str, object],
    after_cache: Mapping[str, object],
    *,
    capability_key: str = "",
    min_score: float = 5.0,
    unique_margin: float = 0.75,
) -> dict[str, object]:
    candidates = []
    objects = after_cache.get("objects")
    if not isinstance(objects, list):
        return {"status": "none", "message": "New SCDM cache does not contain objects.", "candidates": []}

    for raw_object in objects:
        if not isinstance(raw_object, Mapping):
            continue
        signature = raw_object.get("geometrySignature")
        if not isinstance(signature, Mapping):
            continue
        score = _signature_score(before_signature, signature, capability_key=capability_key)
        if score <= 0:
            continue
        candidates.append({"score": score, "object": dict(raw_object), "geometrySignature": dict(signature)})

    candidates.sort(key=lambda item: float(item.get("score") or 0.0), reverse=True)
    if not candidates or float(candidates[0].get("score") or 0.0) < min_score:
        return {"status": "none", "message": "No matching SCDM object reached the confidence threshold.", "candidates": candidates[:5]}
    if len(candidates) > 1:
        top = float(candidates[0].get("score") or 0.0)
        second = float(candidates[1].get("score") or 0.0)
        if top - second < unique_margin:
            return {"status": "multiple", "message": "More than one SCDM object matches the old signature.", "candidates": candidates[:5]}
    best = candidates[0]
    return {
        "status": "unique",
        "message": "Matched one SCDM object.",
        "score": best.get("score"),
        "object": best.get("object"),
        "candidates": candidates[:5],
    }


def build_scdm_id_mapping(
    before_cache: Mapping[str, object],
    after_cache: Mapping[str, object],
    *,
    capability_key: str = "",
) -> dict[str, object]:
    face_id_map: dict[int, int] = {}
    edge_id_map: dict[int, int] = {}
    object_id_map: dict[str, str] = {}
    unmatched: list[str] = []
    ambiguous: list[str] = []

    before_objects = before_cache.get("objects")
    if not isinstance(before_objects, list):
        before_objects = []
    for raw_object in before_objects:
        if not isinstance(raw_object, Mapping):
            continue
        before_signature = raw_object.get("geometrySignature")
        if not isinstance(before_signature, Mapping):
            continue
        object_id = str(raw_object.get("objectId") or "")
        match = match_scdm_object_by_signature(before_signature, after_cache, capability_key=capability_key)
        status = str(match.get("status") or "")
        if status != "unique":
            if status == "multiple":
                ambiguous.append(object_id)
            else:
                unmatched.append(object_id)
            continue
        new_object = match.get("object")
        if not isinstance(new_object, Mapping):
            unmatched.append(object_id)
            continue
        new_signature = new_object.get("geometrySignature")
        if not isinstance(new_signature, Mapping):
            unmatched.append(object_id)
            continue
        new_object_id = str(new_object.get("objectId") or "")
        if object_id and new_object_id:
            object_id_map[object_id] = new_object_id
        _extend_single_or_zipped_id_map(face_id_map, _int_values(before_signature.get("faceIds")), _int_values(new_signature.get("faceIds")))
        _extend_single_or_zipped_id_map(edge_id_map, _int_values(before_signature.get("edgeIds")), _int_values(new_signature.get("edgeIds")))

    return {
        "ok": not unmatched and not ambiguous,
        "objectIdMap": object_id_map,
        "faceIdMap": face_id_map,
        "edgeIdMap": edge_id_map,
        "unmatched": unmatched,
        "ambiguous": ambiguous,
    }


def check_scdm_unedited_objects(
    before_cache: Mapping[str, object],
    after_cache: Mapping[str, object],
    *,
    edited_object_id: str = "",
    edited_signature: Mapping[str, object] | None = None,
    capability_key: str = "",
    max_report: int = 5,
) -> dict[str, object]:
    before_objects = before_cache.get("objects")
    if not isinstance(before_objects, list):
        return {"ok": False, "reason": "missing-before-cache", "message": "Old SCDM cache does not contain objects.", "checked": 0}
    after_objects = after_cache.get("objects")
    if not isinstance(after_objects, list):
        return {"ok": False, "reason": "missing-after-cache", "message": "New SCDM cache does not contain objects.", "checked": 0}

    unmatched: list[dict[str, object]] = []
    ambiguous: list[dict[str, object]] = []
    checked = 0
    for raw_object in before_objects:
        if not isinstance(raw_object, Mapping):
            continue
        object_id = str(raw_object.get("objectId") or "")
        signature = raw_object.get("geometrySignature")
        if not isinstance(signature, Mapping) or not _signature_has_enough_identity(signature):
            continue
        if object_id and edited_object_id and object_id == edited_object_id:
            continue
        if edited_signature and _same_signature_subject(signature, edited_signature):
            continue
        if edited_signature and _is_expected_pattern_spacing_subject(signature, edited_signature, capability_key=capability_key):
            continue
        checked += 1
        match = match_scdm_object_by_signature(signature, after_cache, capability_key=capability_key)
        status = str(match.get("status") or "")
        if status == "unique":
            matched_object = match.get("object")
            matched_signature = matched_object.get("geometrySignature") if isinstance(matched_object, Mapping) else None
            if isinstance(matched_signature, Mapping) and _unchanged_signature_still_matches(signature, matched_signature):
                continue
            status = "none"
        row = {
            "objectId": object_id,
            "objectType": raw_object.get("objectType"),
            "status": status or "none",
            "message": match.get("message"),
        }
        if status == "multiple":
            ambiguous.append(row)
        else:
            unmatched.append(row)

    if unmatched or ambiguous:
        parts = []
        if unmatched:
            parts.append(f"{len(unmatched)} recognized object(s) disappeared or changed too much")
        if ambiguous:
            parts.append(f"{len(ambiguous)} recognized object(s) became ambiguous")
        return {
            "ok": False,
            "reason": "unexpected-object-drift",
            "message": "; ".join(parts) + ".",
            "checked": checked,
            "unmatched": unmatched[:max_report],
            "ambiguous": ambiguous[:max_report],
        }
    return {
        "ok": True,
        "reason": "ok",
        "message": "Unedited recognized objects still match after the SCDM edit.",
        "checked": checked,
        "unmatched": [],
        "ambiguous": [],
    }


def rewrite_scdm_relation_formula_ids(
    text: str,
    mapping: Mapping[str, object],
) -> str:
    return rewrite_relation_formula_ids(
        text,
        _int_map(mapping.get("faceIdMap")),
        _int_map(mapping.get("edgeIdMap")),
    )


def check_scdm_target(
    raw_object: Mapping[str, object],
    *,
    capability_key: str,
    expected_target: object,
    tolerance: float = 1.0e-6,
) -> dict[str, object]:
    if capability_key == "hole.diameter":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "diameter"))
        if actual is None:
            radius = _number(_geometry_value(raw_object, "radius"))
            actual = radius * 2.0 if radius is not None else None
        expected = _number(expected_target)
        return _number_check(actual, expected, "hole.diameter", tolerance)
    if capability_key == "hole.position":
        actual_vector = _vector(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "center"))
        expected_vector = _vector(expected_target)
        return _vector_check(actual_vector, expected_vector, "hole.position", tolerance)
    if capability_key in {"slot.position", "boss.position", "pattern.instance_position"}:
        actual_vector = _vector(
            _capability_value(raw_object, capability_key)
            or _geometry_value(raw_object, "center")
            or _geometry_value(raw_object, "axisCenter")
            or _geometry_value(raw_object, "instanceCenter")
        )
        expected_vector = _vector(expected_target)
        return _vector_check(actual_vector, expected_vector, capability_key, tolerance)
    if capability_key == "slot.width":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "width"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "slot.width", tolerance)
    if capability_key == "slot.depth":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "depth"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "slot.depth", tolerance)
    if capability_key == "boss.height":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "height"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "boss.height", tolerance)
    if capability_key == "boss.diameter":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "diameter"))
        if actual is None:
            radius = _number(_geometry_value(raw_object, "radius"))
            actual = radius * 2.0 if radius is not None else None
        expected = _number(expected_target)
        return _number_check(actual, expected, "boss.diameter", tolerance)
    if capability_key == "round.radius":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "radius"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "round.radius", tolerance)
    if capability_key == "chamfer.distance":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "distance") or _geometry_value(raw_object, "offset"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "chamfer.distance", tolerance)
    if capability_key == "pattern.spacing":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "spacing") or _geometry_value(raw_object, "pitch"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "pattern.spacing", tolerance)
    if capability_key == "pattern.segment_spacing":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "segmentSpacing") or _geometry_value(raw_object, "spacing") or _geometry_value(raw_object, "pitch"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "pattern.segment_spacing", tolerance)
    if capability_key == "shell.thickness":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "thickness"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "shell.thickness", tolerance)
    if capability_key == "face.offset":
        actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "offset") or _geometry_value(raw_object, "planeOffset"))
        expected = _number(expected_target)
        return _number_check(actual, expected, "face.offset", tolerance)
    if _is_removal_capability(capability_key):
        return {"ok": True, "reason": "not-applicable", "message": f"{capability_key} is checked by object disappearance in the caller."}
    return {"ok": None, "reason": "unsupported-post-check", "message": f"No target checker is registered for {capability_key}."}


def _check_pattern_segment_spacing_edit_result(
    edit_result: Mapping[str, object],
    expected_target: object,
    *,
    before_signature: Mapping[str, object] | None,
    tolerance: float,
) -> dict[str, object]:
    applied = edit_result.get("applied")
    nested = edit_result.get("result")
    if not isinstance(applied, Mapping) and isinstance(nested, Mapping):
        applied = nested.get("applied")
    if not isinstance(applied, Mapping):
        return {
            "ok": False,
            "reason": "missing-edit-applied",
            "message": "SCDM edit result did not report the applied local segment spacing.",
        }
    actual = _number(applied.get("segmentSpacing") or applied.get("targetSpacing"))
    expected = _number(expected_target)
    check = _number_check(actual, expected, "pattern.segment_spacing", tolerance)
    if check.get("ok") is True:
        spacing_mode = str(applied.get("spacingMode") or "").strip()
        known_modes = {
            "segment_after",
            "segment_before",
            "segment_split",
            "segment_single_left",
            "segment_single_right",
        }
        if spacing_mode not in known_modes:
            return {
                "ok": False,
                "reason": "pattern-segment-spacing-mode-missing",
                "message": "SCDM edit result did not report a known local spacing mode.",
                "spacingMode": spacing_mode,
                "knownModes": sorted(known_modes),
            }
        expected_modes = _expected_pattern_segment_spacing_modes(before_signature)
        if expected_modes and spacing_mode not in expected_modes:
            return {
                "ok": False,
                "reason": "pattern-segment-spacing-mode-mismatch",
                "message": f"SCDM edit used {spacing_mode}, but the selected modeling intent expected one of {sorted(expected_modes)}.",
                "spacingMode": spacing_mode,
                "expectedModes": sorted(expected_modes),
            }
        check["segmentIndex"] = applied.get("segmentIndex")
        check["spacingMode"] = spacing_mode
    return check


def _expected_pattern_segment_spacing_modes(signature: Mapping[str, object] | None) -> set[str]:
    if not isinstance(signature, Mapping):
        return set()
    moving_side = str(signature.get("movingSide") or "").strip().lower()
    if moving_side in {"after", "right"}:
        return {"segment_after"}
    if moving_side in {"before", "left"}:
        return {"segment_before"}
    if moving_side in {"split", "both", "center"}:
        return {"segment_split"}
    if moving_side in {"single_left", "only_left"}:
        return {"segment_single_left"}
    if moving_side in {"single_right", "only_right"}:
        return {"segment_single_right"}
    return set()


def check_scdm_summary_delta(
    before_cache: Mapping[str, object],
    after_cache: Mapping[str, object],
    *,
    capability_key: str = "",
    relative_tolerance: float = 0.25,
    absolute_tolerance: int = 12,
) -> dict[str, object]:
    if capability_key in {"feature.fill", "feature.delete_round_or_chamfer"}:
        return {"ok": None, "reason": "skipped-command-feature", "message": "Command features are expected to change Face/Edge counts."}
    before_summary = _raw_summary(before_cache)
    after_summary = _raw_summary(after_cache)
    if not before_summary or not after_summary:
        return {"ok": None, "reason": "missing-summary", "message": "SCDM raw summary was not available in both caches."}

    body_before = _int_or_none(before_summary.get("bodyCount"))
    body_after = _int_or_none(after_summary.get("bodyCount"))
    if body_before is not None and body_after is not None and body_before != body_after:
        return {
            "ok": None,
            "reason": "body-count-repartitioned",
            "message": f"SCDM result changed body count during STEP re-export: {body_before} -> {body_after}.",
            "before": dict(before_summary),
            "after": dict(after_summary),
            "metric": "bodyCount",
            "beforeValue": body_before,
            "afterValue": body_after,
        }

    for key, label in (("faceCount", "Face"), ("edgeCount", "Edge"), ("objectCount", "对象")):
        before_value = _int_or_none(before_summary.get(key))
        after_value = _int_or_none(after_summary.get(key))
        if before_value is None or after_value is None:
            continue
        delta = abs(after_value - before_value)
        limit = max(int(absolute_tolerance), int(math.ceil(abs(before_value) * float(relative_tolerance))))
        if delta > limit:
            return {
                "ok": False,
                "reason": "summary-drift",
                "message": f"SCDM result changed {label} count too much: {before_value} -> {after_value}, limit {limit}.",
                "before": dict(before_summary),
                "after": dict(after_summary),
                "metric": key,
                "delta": delta,
                "limit": limit,
            }

    return {
        "ok": True,
        "reason": "ok",
        "message": "SCDM model summary stayed within the allowed range.",
        "before": dict(before_summary),
        "after": dict(after_summary),
    }


def _signature_score(before: Mapping[str, object], after: Mapping[str, object], *, capability_key: str) -> float:
    score = 0.0
    before_type = str(before.get("objectType") or "").lower()
    after_type = str(after.get("objectType") or "").lower()
    if before_type and before_type == after_type:
        score += 4.0
    elif {before_type, after_type} <= {"hole", "cylindrical_hole", ""}:
        score += 2.0

    before_surface = str(before.get("surfaceType") or "").lower()
    after_surface = str(after.get("surfaceType") or "").lower()
    if before_surface and before_surface == after_surface:
        score += 1.0

    before_components = _component_locator_keys(before.get("componentLocators") or before.get("bodyLocators"))
    after_components = _component_locator_keys(after.get("componentLocators") or after.get("bodyLocators"))
    if before_components and after_components:
        if before_components & after_components:
            score += 3.0
        else:
            return 0.0

    before_body = _int_or_none(before.get("bodyIndex"))
    after_body = _int_or_none(after.get("bodyIndex"))
    if before_body is not None and after_body is not None and before_body == after_body:
        score += 2.0
        before_face_ordinal = _int_or_none(before.get("faceOrdinal"))
        after_face_ordinal = _int_or_none(after.get("faceOrdinal"))
        if before_face_ordinal is not None and before_face_ordinal == after_face_ordinal:
            score += 1.0
        before_edge_ordinal = _int_or_none(before.get("edgeOrdinal"))
        after_edge_ordinal = _int_or_none(after.get("edgeOrdinal"))
        if before_edge_ordinal is not None and before_edge_ordinal == after_edge_ordinal:
            score += 1.0

    before_faces = set(_int_values(before.get("faceIds")))
    after_faces = set(_int_values(after.get("faceIds")))
    if before_faces and after_faces:
        overlap = len(before_faces & after_faces)
        if overlap:
            score += 1.0 + min(overlap, 3) * 0.25

    before_edges = set(_int_values(before.get("edgeIds")))
    after_edges = set(_int_values(after.get("edgeIds")))
    if before_edges and after_edges and before_edges & after_edges:
        score += 0.5

    if not _is_position_capability(capability_key):
        before_center = _vector(before.get("center"))
        after_center = _vector(after.get("center"))
        if _is_removal_capability(capability_key) and before_center and after_center:
            if _vector_error(before_center, after_center) > _vector_tolerance(before_center, after_center):
                return 0.0
        center_score = _vector_distance_score(before_center, after_center)
        score += center_score

    axis_score = _axis_score(_vector(before.get("axis")), _vector(after.get("axis")))
    score += axis_score

    if capability_key != "hole.diameter":
        score += _number_similarity_score(_diameter_from_signature(before), _diameter_from_signature(after))

    return score


def _is_position_capability(capability_key: str) -> bool:
    return capability_key in {"hole.position", "slot.position", "boss.position", "pattern.instance_position"} or capability_key.endswith(".position")


def _is_removal_capability(capability_key: str) -> bool:
    return capability_key in {"feature.fill", "feature.delete_round_or_chamfer"} or capability_key.endswith(".remove") or capability_key.startswith("feature.delete")


def _check_removed_object_match(match: Mapping[str, object]) -> dict[str, object]:
    status = str(match.get("status") or "")
    if status == "none":
        return {"ok": True, "reason": "removed", "message": "Edited feature is no longer present in the new SCDM cache."}
    if status == "unique":
        return {
            "ok": False,
            "reason": "feature-still-present",
            "message": "SCDM reported success, but the edited feature still matches an object in the new cache.",
            "match": dict(match),
        }
    if status == "multiple":
        return {
            "ok": False,
            "reason": "feature-removal-ambiguous",
            "message": "SCDM reported success, but multiple new objects still match the edited feature.",
            "match": dict(match),
        }
    return {
        "ok": False,
        "reason": f"feature-removal-{status or 'failed'}",
        "message": str(match.get("message") or "Removed feature could not be verified."),
        "match": dict(match),
    }


def _is_expected_pattern_spacing_subject(
    signature: Mapping[str, object],
    edited_signature: Mapping[str, object],
    *,
    capability_key: str,
) -> bool:
    if capability_key not in {"pattern.spacing", "pattern.segment_spacing"}:
        return False
    if str(edited_signature.get("objectType") or "") != "linear_pattern":
        return False
    touched_faces = set(_int_values(edited_signature.get("faceIds")))
    touched_bodies = set(_int_values(edited_signature.get("bodyIndices")))
    touched_components = _component_locator_keys(edited_signature.get("componentLocators"))
    for instance in _pattern_instances(edited_signature):
        touched_faces.update(_int_values(instance.get("faceIds")))
        body_index = _int_or_none(instance.get("bodyIndex"))
        if body_index is not None:
            touched_bodies.add(body_index)
        touched_bodies.update(_int_values(instance.get("bodyIndices")))
        touched_components.update(_component_locator_keys(instance.get("componentLocators") or instance.get("bodyLocators")))

    subject_faces = set(_int_values(signature.get("faceIds")))
    if touched_faces and subject_faces and touched_faces.intersection(subject_faces):
        return True
    subject_body = _int_or_none(signature.get("bodyIndex"))
    if subject_body is not None and subject_body in touched_bodies:
        return True
    subject_bodies = set(_int_values(signature.get("bodyIndices")))
    if touched_bodies and subject_bodies and touched_bodies.intersection(subject_bodies):
        return True
    subject_components = _component_locator_keys(signature.get("componentLocators") or signature.get("bodyLocators"))
    return bool(touched_components and subject_components and touched_components.intersection(subject_components))


def _pattern_instances(signature: Mapping[str, object]) -> list[Mapping[str, object]]:
    value = signature.get("patternInstances")
    if not isinstance(value, (list, tuple)):
        return []
    return [item for item in value if isinstance(item, Mapping)]


def _component_locator_keys(value: object) -> set[str]:
    if not isinstance(value, (list, tuple)):
        return set()
    result: set[str] = set()
    for locator in value:
        if not isinstance(locator, Mapping):
            continue
        path = _ordered_int_values(locator.get("componentPath"))
        if path:
            result.add("path:" + ".".join(str(item) for item in path))
            continue
        component_index = _int_or_none(locator.get("componentIndex"))
        if component_index is not None:
            result.add("index:" + str(component_index))
    return result


def _signature_has_enough_identity(signature: Mapping[str, object]) -> bool:
    if _vector(signature.get("center")) and _vector(signature.get("axis")):
        return True
    if _vector(signature.get("center")) and str(signature.get("surfaceType") or ""):
        return True
    if _int_values(signature.get("faceIds")) or _int_values(signature.get("edgeIds")):
        return True
    return False


def _same_signature_subject(left: Mapping[str, object], right: Mapping[str, object]) -> bool:
    left_faces = set(_int_values(left.get("faceIds")))
    right_faces = set(_int_values(right.get("faceIds")))
    if left_faces and right_faces and left_faces == right_faces:
        return True
    left_edges = set(_int_values(left.get("edgeIds")))
    right_edges = set(_int_values(right.get("edgeIds")))
    if left_edges and right_edges and left_edges == right_edges:
        return True
    left_center = _vector(left.get("center"))
    right_center = _vector(right.get("center"))
    if len(left_center) == 3 and len(right_center) == 3 and _vector_error(left_center, right_center) <= 1.0e-8:
        left_axis = _vector(left.get("axis"))
        right_axis = _vector(right.get("axis"))
        if len(left_axis) == 3 and len(right_axis) == 3 and _axis_score(left_axis, right_axis) >= 3.0:
            return True
    return False


def _unchanged_signature_still_matches(before: Mapping[str, object], after: Mapping[str, object]) -> bool:
    before_center = _vector(before.get("center"))
    after_center = _vector(after.get("center"))
    if before_center and after_center and _vector_error(before_center, after_center) > _vector_tolerance(before_center, after_center):
        return False

    before_axis = _vector(before.get("axis"))
    after_axis = _vector(after.get("axis"))
    if before_axis and after_axis and _axis_score(before_axis, after_axis) < 3.0:
        return False

    before_diameter = _diameter_from_signature(before)
    after_diameter = _diameter_from_signature(after)
    if before_diameter is not None and after_diameter is not None:
        tolerance = max(abs(before_diameter), abs(after_diameter), 1.0) * 1.0e-5
        if abs(float(before_diameter) - float(after_diameter)) > tolerance:
            return False

    for key in ("planeOffset", "width", "depth", "height", "distance", "spacing", "pitch", "thickness"):
        before_value = _number(before.get(key))
        after_value = _number(after.get(key))
        if before_value is not None and after_value is not None:
            tolerance = max(abs(before_value), abs(after_value), 1.0) * 1.0e-5
            if abs(float(before_value) - float(after_value)) > tolerance:
                return False
    return True


def _vector_tolerance(left: Sequence[float], right: Sequence[float]) -> float:
    scale = 1.0
    values = list(left) + list(right)
    if values:
        scale = max(scale, max(abs(float(item)) for item in values))
    return max(scale * 1.0e-5, 1.0e-7)


def _extend_single_or_zipped_id_map(target: dict[int, int], old_ids: Sequence[int], new_ids: Sequence[int]) -> None:
    old_unique = sorted(set(old_ids))
    new_unique = sorted(set(new_ids))
    if len(old_unique) == 1 and len(new_unique) == 1:
        target[int(old_unique[0])] = int(new_unique[0])
    elif len(old_unique) == len(new_unique) and len(old_unique) > 1:
        for old_id, new_id in zip(old_unique, new_unique):
            target[int(old_id)] = int(new_id)


def _capability_value(raw_object: Mapping[str, object], capability_key: str) -> object:
    capabilities = raw_object.get("capabilities")
    if not isinstance(capabilities, list):
        return None
    for capability in capabilities:
        if isinstance(capability, Mapping) and capability.get("key") == capability_key:
            return capability.get("currentValue")
    return None


def _geometry_value(raw_object: Mapping[str, object], key: str) -> object:
    signature = raw_object.get("geometrySignature")
    if isinstance(signature, Mapping) and key in signature:
        return signature.get(key)
    geometry = raw_object.get("geometry")
    if isinstance(geometry, Mapping) and key in geometry:
        return geometry.get(key)
    return None


def _number_check(actual: float | None, expected: float | None, label: str, tolerance: float) -> dict[str, object]:
    if actual is None or expected is None:
        return {"ok": False, "reason": "target-value-missing", "message": f"{label} target check does not have comparable values.", "actual": actual, "expected": expected}
    error = abs(actual - expected)
    return {
        "ok": error <= tolerance,
        "reason": "ok" if error <= tolerance else "target-mismatch",
        "message": "Target value matched." if error <= tolerance else f"{label} actual={actual:g}, expected={expected:g}, error={error:g}.",
        "actual": actual,
        "expected": expected,
        "error": error,
        "tolerance": tolerance,
    }


def _vector_check(actual: Sequence[float], expected: Sequence[float], label: str, tolerance: float) -> dict[str, object]:
    if len(actual) != 3 or len(expected) != 3:
        return {"ok": False, "reason": "target-value-missing", "message": f"{label} target check does not have comparable vectors.", "actual": list(actual), "expected": list(expected)}
    error = math.sqrt(sum((float(actual[index]) - float(expected[index])) ** 2 for index in range(3)))
    return {
        "ok": error <= tolerance,
        "reason": "ok" if error <= tolerance else "target-mismatch",
        "message": "Target vector matched." if error <= tolerance else f"{label} vector error={error:g}.",
        "actual": list(actual),
        "expected": list(expected),
        "error": error,
        "tolerance": tolerance,
    }


def _vector_error(actual: Sequence[float], expected: Sequence[float]) -> float:
    if len(actual) != 3 or len(expected) != 3:
        return math.inf
    return math.sqrt(sum((float(actual[index]) - float(expected[index])) ** 2 for index in range(3)))


def _vector_distance_score(before: Sequence[float], after: Sequence[float]) -> float:
    if len(before) != 3 or len(after) != 3:
        return 0.0
    distance = math.sqrt(sum((float(before[index]) - float(after[index])) ** 2 for index in range(3)))
    if distance <= 1.0e-5:
        return 4.0
    if distance <= 1.0e-3:
        return 3.0
    if distance <= 1.0e-1:
        return 1.5
    return 0.0


def _axis_score(before: Sequence[float], after: Sequence[float]) -> float:
    if len(before) != 3 or len(after) != 3:
        return 0.0
    before_len = math.sqrt(sum(float(item) * float(item) for item in before))
    after_len = math.sqrt(sum(float(item) * float(item) for item in after))
    if before_len <= 1.0e-12 or after_len <= 1.0e-12:
        return 0.0
    dot = abs(sum(float(before[index]) * float(after[index]) for index in range(3)) / (before_len * after_len))
    if dot >= 0.999:
        return 3.0
    if dot >= 0.99:
        return 2.0
    return 0.0


def _number_similarity_score(before: float | None, after: float | None) -> float:
    if before is None or after is None:
        return 0.0
    error = abs(float(before) - float(after))
    scale = max(abs(float(before)), abs(float(after)), 1.0)
    relative = error / scale
    if relative <= 1.0e-5:
        return 3.0
    if relative <= 1.0e-3:
        return 2.0
    if relative <= 5.0e-2:
        return 0.75
    return 0.0


def _diameter_from_signature(signature: Mapping[str, object]) -> float | None:
    diameter = _number(signature.get("diameter"))
    if diameter is not None:
        return diameter
    radius = _number(signature.get("radius"))
    return radius * 2.0 if radius is not None else None


def _number(value: object) -> float | None:
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _vector(value: object) -> list[float]:
    if isinstance(value, (str, bytes)) or value is None:
        return []
    try:
        values = list(value)  # type: ignore[arg-type]
    except TypeError:
        return []
    if len(values) != 3:
        return []
    try:
        return [float(item) for item in values]
    except (TypeError, ValueError):
        return []


def _int_values(value: object) -> list[int]:
    if isinstance(value, (str, bytes)) or value is None:
        return []
    try:
        values = list(value)  # type: ignore[arg-type]
    except TypeError:
        return []
    result: list[int] = []
    for item in values:
        try:
            result.append(int(item))
        except (TypeError, ValueError):
            continue
    return result


def _ordered_int_values(value: object) -> list[int]:
    if isinstance(value, (str, bytes)) or value is None:
        return []
    try:
        values = list(value)  # type: ignore[arg-type]
    except TypeError:
        return []
    result: list[int] = []
    for item in values:
        try:
            result.append(int(item))
        except (TypeError, ValueError):
            continue
    return result


def _raw_summary(cache: Mapping[str, object]) -> Mapping[str, object]:
    diagnostics = cache.get("diagnostics")
    if not isinstance(diagnostics, Mapping):
        return {}
    summary = diagnostics.get("raw_summary") or diagnostics.get("summary")
    return summary if isinstance(summary, Mapping) else {}


def _int_or_none(value: object) -> int | None:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _int_map(value: object) -> dict[int, int]:
    if not isinstance(value, Mapping):
        return {}
    result: dict[int, int] = {}
    for key, item in value.items():
        try:
            result[int(key)] = int(item)
        except (TypeError, ValueError):
            continue
    return result


__all__ = [
    "build_scdm_id_mapping",
    "check_scdm_summary_delta",
    "check_scdm_unedited_objects",
    "check_scdm_target",
    "match_scdm_object_by_signature",
    "rewrite_scdm_relation_formula_ids",
    "validate_scdm_edit_result",
]