from __future__ import annotations import importlib.util import json import math import tempfile import unittest from pathlib import Path SCRIPT = ( Path(__file__).parents[1] / "skills" / "cad-experience-builder" / "scripts" / "cad_experience.py" ) SPEC = importlib.util.spec_from_file_location("cad_experience", SCRIPT) MODULE = importlib.util.module_from_spec(SPEC) assert SPEC.loader SPEC.loader.exec_module(MODULE) STEP_SCRIPT = ( Path(__file__).parents[1] / "skills" / "cad-experience-builder" / "scripts" / "step_to_case.py" ) STEP_SPEC = importlib.util.spec_from_file_location("step_to_case", STEP_SCRIPT) STEP_MODULE = importlib.util.module_from_spec(STEP_SPEC) assert STEP_SPEC.loader STEP_SPEC.loader.exec_module(STEP_MODULE) def make_case(index: int) -> dict: return { "provenance": {"source_sha256": f"case_{index}"}, "design_ir": { "part_family": "flanged_hub_adapter", "features": [ {"type": "base_flange", "center": [91.0, 52.0]}, {"type": "hollow_sleeve", "surface_ids": ["face_1"]}, {"type": "counterbore"}, ], "constraints": [{"id": "coaxial_stack", "type": "coaxial"}], "parameters": { "base_outer_diameter": 91.0, "counterbore_spacing": 52.0, }, "normalized_observations": [ { "name": "sleeve_od_to_base_od", "value": 0.3 + index * 0.001, "numerator_role": "sleeve_outer_diameter", "denominator_role": "base_outer_diameter", } ], "reconstruction_evidence": { "parameter_roles": [ "overall_long_span", "overall_middle_span", "overall_short_span", ], "datum_roles": ["part_center", "primary_axis"], "feature_roles": ["base_flange", "hollow_sleeve", "counterbore"], "relation_roles": ["coaxial_stack"], "canonical_stages": [ { "id": "establish_reference_frame", "operation": "define_datums", "feature_roles": [], "reference_roles": ["part_center", "primary_axis"], }, { "id": "construct_primary_envelope", "operation": "revolve_profile", "feature_roles": ["base_flange", "hollow_sleeve"], "reference_roles": ["primary_axis"], }, ], "private_cardinality_evidence": { "surface_type_classes": { "plane": "repeated", "cylinder": "dense", }, "feature_role_count_class": "repeated", }, "validation_roles": ["closed_solid", "coaxial_stack"], }, }, "experience": { "rules": [ { "id": "bound_counterbore_to_host", "scope": ["flanged_hub_adapter"], "statement": "Terminate a counterbore at its host boundary.", "repair": "Bind the cutter extent to the host feature.", } ], "validation_targets": [ { "id": "host_boundary", "check": "A counterbore does not cross an unrelated adjacent feature.", } ], }, } def positive_validation(*experience_ids: str) -> dict: return { "validation_kind": "heldout_experience_ab", "method_results": [ { "experience_id": experience_id, "independent_case_count": 3, "edit_contract_pass_rate": 1.0, "baseline_score_delta": 0.5, "passed": True, } for experience_id in experience_ids ], } class CadExperienceTest(unittest.TestCase): def test_default_data_paths_use_shallow_local_layout(self): pipeline_root = Path(__file__).parents[1].resolve() self.assertEqual(pipeline_root / "input", MODULE.input_root()) self.assertEqual(pipeline_root / "runs", MODULE.runs_root()) self.assertEqual( pipeline_root / "output" / "library.json", MODULE.default_library_path(), ) def reviewed_proposal(self) -> dict: return { "id": "constraint.flanged_hub_adapter.shared_axis", "kind": "constraint", "scope": ["flanged_hub_adapter"], "evidence_query": { "required_features": ["base_flange", "hollow_sleeve"], "required_relations": ["coaxial_stack"], }, "guidance": "Preserve a shared axis across participating feature roles.", "semantic_rationale": "The relationship transfers across dimensional variants.", } def test_semantic_summary_derives_roles_without_copying_dimensions(self): surfaces = [ { "type": "cylinder", "axis": [0.0, 1.0, 0.0], "location": [0.0, 0.0, 0.0], "radius": radius, "area": 100.0, } for radius in (5.0, 10.0, 20.0) ] surfaces.extend( { "type": "cylinder", "axis": [0.0, 1.0, 0.0], "location": [offset, 0.0, 0.0], "radius": 2.0, "area": 20.0, } for offset in (-15.0, 15.0, 16.0) ) surfaces.append( { "type": "cone", "axis": [0.0, 1.0, 0.0], "location": [0.0, 0.0, 0.0], "reference_radius": 10.0, "semi_angle_radians": 0.5, "area": 50.0, } ) family, features, constraints, summary = STEP_MODULE._semantic_summary( surfaces, [0.0, 0.0, 0.0], [40.0, 20.0, 40.0] ) self.assertEqual("flanged_rotational_part", family) self.assertIn( "coaxial_cylindrical_stack", {item["type"] for item in features} ) self.assertIn("coaxial_stack", {item["id"] for item in constraints}) self.assertEqual("y", summary["dominant_axis"]) self.assertTrue(summary["normalized_observations"]) reconstruction = STEP_MODULE._reconstruction_evidence( surfaces, features, constraints, summary ) self.assertIn("overall_long_span", reconstruction["parameter_roles"]) self.assertTrue(reconstruction["canonical_stages"]) self.assertEqual( "canonical_reconstruction_plan_not_recovered_history", reconstruction["interpretation"], ) parameters = STEP_MODULE._semantic_parameter_candidates( surfaces, [0.0, 0.0, 0.0], [40.0, 20.0, 40.0], features, summary, ) names = {item["name"] for item in parameters} self.assertIn("repeated_feature_count", names) self.assertIn("repeated_feature_diameter", names) self.assertTrue( all( item["epistemic_status"] == "inferred_from_final_brep" for item in parameters ) ) def test_non_solid_teacher_is_rejected_from_induction(self): payload = make_case(1) payload["geometry_evidence"] = { "valid": False, "reconstruction_eligible": False, "solid_count": 0, "volume": 0.0, "rejection_reasons": ["no_closed_solid", "zero_volume"], } normalized = MODULE.normalize_case(payload) self.assertFalse(normalized["reconstruction_eligible"]) self.assertIn("no_closed_solid", normalized["rejection_reasons"]) def test_spur_gear_flanks_are_not_mislabeled_as_hole_pattern(self): surfaces = [ { "type": "cylinder", "axis": [1.0, 0.0, 0.0], "location": [0.0, 0.0, 0.0], "radius": radius, "area": 100.0, } for radius in (3.0, 8.8) ] for index in range(40): angle = 2.0 * math.pi * index / 40.0 surfaces.append( { "type": "cylinder", "axis": [1.0, 0.0, 0.0], "location": [ 0.0, 7.5 * math.cos(angle), 7.5 * math.sin(angle), ], "radius": 3.2, "area": 10.0, } ) surfaces.extend( {"type": "plane", "area": 250.0} for _ in range(2) ) family, features, constraints, summary = STEP_MODULE._semantic_summary( surfaces, [4.0, 0.0, 0.0], [8.0, 17.6, 17.6] ) feature_roles = {item["type"] for item in features} self.assertEqual("external_spur_gear", family) self.assertEqual(20, summary["inferred_tooth_count"]) self.assertIn("external_spur_gear", feature_roles) self.assertIn("equal_angular_tooth_pattern", feature_roles) self.assertNotIn("repeated_axial_hole_pattern", feature_roles) self.assertIn( "equal_angular_tooth_pattern", {item["id"] for item in constraints}, ) parameters = STEP_MODULE._semantic_parameter_candidates( surfaces, [4.0, 0.0, 0.0], [8.0, 17.6, 17.6], features, summary, ) values = {item["name"]: item["value"] for item in parameters} self.assertEqual(20, values["tooth_count"]) self.assertAlmostEqual(0.8, values["gear_module"], places=6) self.assertEqual(8.0, values["gear_width"]) self.assertNotIn("repeated_feature_count", values) def test_sha256_identity_deduplicates_renamed_sources(self): digest = "0" * 64 with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) for index in range(2): payload = make_case(index) payload["provenance"]["source_sha256"] = digest (root / f"renamed-{index}.json").write_text(json.dumps(payload)) cases, stats = MODULE.load_cases(root) self.assertEqual(1, len(cases)) self.assertEqual(1, stats["duplicate_case_count"]) def test_conditional_feature_context_does_not_dilute_valid_method(self): proposal = self.reviewed_proposal() proposal["evidence_query"]["context_features"] = [ "base_flange", "hollow_sleeve", ] target = [MODULE.normalize_case(make_case(index)) for index in range(20)] unrelated = [] for index in range(80): payload = make_case(index + 100) payload["design_ir"]["features"] = [{"type": "base_flange"}] payload["design_ir"]["constraints"] = [] unrelated.append(MODULE.normalize_case(payload)) library = MODULE.build_reviewed_library( target + unrelated, [MODULE.normalize_proposals({ "draft_kind": "llm_generalized_experience_proposals", "proposals": [proposal], })[0]], 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, positive_validation(proposal["id"]), ) self.assertEqual(1, len(library["experiences"])) def test_single_case_never_promotes(self): case = MODULE.normalize_case(make_case(1)) library = MODULE.induce( [case], 2, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, ) self.assertEqual(library["status"], "collecting_evidence") self.assertEqual(library["experiences"], []) self.assertTrue(library["candidate_experiences"]) self.assertTrue( all( item["promotion_state"] == "candidate" for item in library["candidate_experiences"] ) ) self.assertTrue( all( item["consumer_policy"] == "visible_for_review_but_not_available_to_cad_router" for item in library["candidate_experiences"] ) ) def test_batch_promotes_methods_without_instance_values(self): cases = [MODULE.normalize_case(make_case(index)) for index in range(20)] library = MODULE.induce( cases, 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, ) self.assertEqual(MODULE.audit_library(library), []) rendered = json.dumps(library) self.assertNotIn('"parameters"', rendered) self.assertNotIn("91.0", rendered) self.assertNotIn("52.0", rendered) self.assertNotIn('"center"', rendered) ids = {item["id"] for item in library["experiences"]} self.assertIn("rule.bound_counterbore_to_host", ids) self.assertIn("constraint.flanged_hub_adapter.coaxial_stack", ids) self.assertIn("distribution.sleeve_od_to_base_od", ids) def test_audit_allows_aggregate_validation_counts_only(self): library = { "schema_version": "2.0", "library_kind": "generalized_cad_experience", "experiences": [ { "id": "edit_strategy.test", "support": 20, "validation": { "heldout_pair_count": 20, "treatment_pass_count": 20, "regressed_pair_count": 0, "pass_rate_delta": 0.25, }, } ], "candidate_experiences": [], } self.assertEqual([], MODULE.audit_library(library)) library["experiences"][0]["validation"]["radius"] = 3.0 errors = MODULE.audit_library(library) self.assertTrue(any("forbidden instance key" in error for error in errors)) def test_family_support_is_not_diluted_by_unrelated_families(self): target = [MODULE.normalize_case(make_case(index)) for index in range(20)] unrelated = [] for index in range(100): payload = make_case(index + 1000) payload["design_ir"]["part_family"] = "gear" payload["design_ir"]["features"] = [ {"type": "gear_blank"}, {"type": "gear_teeth"}, ] unrelated.append(MODULE.normalize_case(payload)) library = MODULE.induce( target + unrelated, 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, ) ids = {item["id"] for item in library["experiences"]} self.assertIn( "motif.flanged_hub_adapter.base_flange+hollow_sleeve", ids ) def test_query_returns_backend_neutral_context(self): cases = [MODULE.normalize_case(make_case(index)) for index in range(20)] library = MODULE.induce( cases, 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, ) result = MODULE.query_library( library, "flanged_hub_adapter", {"base_flange", "hollow_sleeve", "counterbore"}, ) self.assertTrue(result["experiences"]) self.assertFalse(result["policy"]["contains_instance_parameters"]) def test_llm_proposal_is_candidate_until_evidence_threshold(self): proposal = self.reviewed_proposal() one_case = MODULE.normalize_case(make_case(1)) candidate_library = MODULE.build_reviewed_library( [one_case], [proposal], 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, ) self.assertEqual([], candidate_library["experiences"]) self.assertEqual(1, len(candidate_library["candidate_experiences"])) self.assertFalse(candidate_library["policy"]["router_consumable"]) cases = [MODULE.normalize_case(make_case(index)) for index in range(20)] promoted_library = MODULE.build_reviewed_library( cases, [proposal], 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, positive_validation(proposal["id"]), ) self.assertEqual(1, len(promoted_library["experiences"])) self.assertEqual([], promoted_library["candidate_experiences"]) def test_parameter_naming_requires_positive_heldout_validation(self): cases = [] for index in range(20): payload = make_case(index) payload["design_ir"]["semantic_parameter_candidates"] = [ { "name": "bolt_circle_diameter", "unit": "mm", "confidence": 0.9, "semantic_aliases": ["hole_spacing"], "affects_feature_roles": ["mounting_pattern"], "epistemic_status": "inferred_from_final_brep", "edit_contract": { "expected_change": "pattern_radius_changes" }, } ] cases.append(MODULE.normalize_case(payload)) raw = { "id": "parameter.flange.bolt_circle_diameter", "kind": "parameter_naming", "scope": ["flanged_hub_adapter"], "evidence_query": { "required_features": ["base_flange"], "required_relations": [], "required_parameters": ["bolt_circle_diameter"], }, "guidance": "Expose the circular mounting pattern diameter as an editable semantic parameter.", "semantic_rationale": "The named role preserves edit intent across dimensional variants.", "canonical_parameter": "bolt_circle_diameter", "semantic_aliases": ["hole_spacing"], } proposal = MODULE.normalize_proposals( { "draft_kind": "llm_generalized_experience_proposals", "proposals": [raw], } )[0] unvalidated = MODULE.build_reviewed_library( cases, [proposal], 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, ) self.assertEqual([], unvalidated["experiences"]) self.assertEqual( "awaiting_positive_heldout_ab", unvalidated["candidate_experiences"][0]["validation_state"], ) validated = MODULE.build_reviewed_library( cases, [proposal], 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, positive_validation(raw["id"]), ) self.assertEqual("parameter_naming", validated["experiences"][0]["kind"]) def test_numeric_instance_answer_is_rejected_from_llm_draft(self): proposal = self.reviewed_proposal() proposal["guidance"] = "Always use a diameter of 91 millimeters." with self.assertRaisesRegex(ValueError, "instance numbers"): MODULE.normalize_proposals( { "draft_kind": "llm_generalized_experience_proposals", "proposals": [proposal], } ) def test_reconstruction_grammar_promotes_without_instance_values(self): proposal = { "id": "reconstruction.flanged_hub_adapter.primary", "kind": "reconstruction_grammar", "scope": ["flanged_hub_adapter"], "evidence_query": { "required_features": ["base_flange", "hollow_sleeve"], "required_relations": ["coaxial_stack"], "context_features": ["base_flange", "hollow_sleeve"], }, "guidance": "Establish shared datums before composing the primary envelope.", "semantic_rationale": "The symbolic stages transfer across dimensional variants.", "reconstruction_grammar": { "parameter_roles": [ "overall_long_span", "overall_middle_span", "overall_short_span", ], "datum_roles": ["part_center", "primary_axis"], "feature_roles": ["base_flange", "hollow_sleeve"], "relation_roles": ["coaxial_stack"], "canonical_stages": [ { "id": "establish_reference_frame", "operation": "define_datums", "feature_roles": [], "reference_roles": ["part_center", "primary_axis"], }, { "id": "construct_primary_envelope", "operation": "revolve_profile", "feature_roles": ["base_flange", "hollow_sleeve"], "reference_roles": ["primary_axis"], }, ], "cardinality_classes": ["plane_repeated", "cylinder_dense"], "validation_roles": ["closed_solid", "coaxial_stack"], }, } normalized = MODULE.normalize_proposals( { "draft_kind": "llm_generalized_experience_proposals", "proposals": [proposal], } ) cases = [MODULE.normalize_case(make_case(index)) for index in range(20)] library = MODULE.build_reviewed_library( cases, normalized, 20, 0.8, {"duplicate_case_count": 0, "rejected_case_count": 0}, positive_validation(proposal["id"]), ) self.assertEqual([], MODULE.audit_library(library)) self.assertEqual("reconstruction_grammar", library["experiences"][0]["kind"]) rendered = json.dumps(library) self.assertNotIn("91.0", rendered) self.assertNotIn('"parameters"', rendered) def test_direct_distill_without_model_draft_is_blocked(self): with self.assertRaisesRegex(SystemExit, "statistical distillation is disabled"): MODULE.main(["distill"]) if __name__ == "__main__": unittest.main()