1543 lines
66 KiB
Python
1543 lines
66 KiB
Python
"""Offline experiment for geometry-only CADFS selector candidates.
|
|
|
|
The production resolver must not consume this module. It removes provenance
|
|
intent only in a copied CDSL document, then compares that diagnostic replay to
|
|
the source's final STEP. A strict passing probe is replayed a second time from
|
|
its bound CDSL and can emit a *behavioural* selector record. The record is
|
|
training/diagnostic evidence, not a claim that a heuristic recovered Onshape's
|
|
hidden topology identity.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from copy import deepcopy
|
|
import json
|
|
import math
|
|
import multiprocessing
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .compare import compare_steps
|
|
from .dataset import Sample, scan_dataset
|
|
from .featurescript_parser import parse_featurescript
|
|
from .lowering import lower_model
|
|
from .rebuild import rebuild_candidate
|
|
from .reports import read_json, write_json, write_manifest
|
|
|
|
|
|
def strip_provenance_intents(value: Any) -> int:
|
|
"""Remove provenance-only selector fields from one copied CDSL value."""
|
|
removed = 0
|
|
if isinstance(value, dict):
|
|
removed_intent = False
|
|
for key in ("selector_intent", "selector_intent_version"):
|
|
if key in value:
|
|
value.pop(key)
|
|
removed += 1
|
|
removed_intent = True
|
|
# Feature output roles are semantic builder evidence, just like a
|
|
# selector intent. The geometry-only probe must not leave an output
|
|
# role in place after discarding the evidence that validates it.
|
|
if removed_intent:
|
|
for key in ("output_role", "output_role_source"):
|
|
if key in value:
|
|
value.pop(key)
|
|
for child in value.values():
|
|
removed += strip_provenance_intents(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
removed += strip_provenance_intents(child)
|
|
return removed
|
|
|
|
|
|
def _write_rebuild_artifacts(directory: Path, name: str, outcome: dict[str, Any]) -> None:
|
|
"""Persist the same useful replay artifacts as the regular pipeline."""
|
|
artifact = deepcopy(outcome)
|
|
bound = artifact.pop("bound_cdsl", None)
|
|
prefix = artifact.get("last_executable_prefix")
|
|
if isinstance(prefix, dict):
|
|
prefix_bound = prefix.pop("bound_cdsl", None)
|
|
if bound is None:
|
|
bound = prefix_bound
|
|
write_json(directory / f"{name}.rebuild.json", artifact)
|
|
if isinstance(bound, dict):
|
|
write_json(directory / f"{name}.bound.cdsl.json", bound)
|
|
|
|
|
|
def _rebuild_worker(candidate: str, step: str, result: str) -> None:
|
|
write_json(Path(result), rebuild_candidate(read_json(Path(candidate)), Path(step)))
|
|
|
|
|
|
def _compare_worker(gold: str, rebuilt: str, result: str) -> None:
|
|
write_json(Path(result), compare_steps(Path(gold), Path(rebuilt)))
|
|
|
|
|
|
def _isolated(target: Any, args: tuple[Any, ...], result_path: Path, timeout_seconds: float) -> str:
|
|
"""Bound an OCC experiment so one candidate cannot stall the demo."""
|
|
result_path.unlink(missing_ok=True)
|
|
process = multiprocessing.get_context("spawn").Process(target=target, args=args)
|
|
process.start()
|
|
process.join(timeout_seconds)
|
|
if process.is_alive():
|
|
process.terminate()
|
|
process.join(5)
|
|
if process.is_alive():
|
|
process.kill()
|
|
process.join()
|
|
return "timeout"
|
|
return "completed" if process.exitcode == 0 and result_path.exists() else "failed"
|
|
|
|
|
|
def _run_rebuild(candidate_path: Path, step: Path, *, timeout_seconds: float) -> dict[str, Any]:
|
|
worker = step.with_suffix(".worker.json")
|
|
outcome = _isolated(_rebuild_worker, (str(candidate_path), str(step), str(worker)), worker, timeout_seconds)
|
|
if outcome == "completed":
|
|
result = read_json(worker)
|
|
worker.unlink(missing_ok=True)
|
|
return result
|
|
step.unlink(missing_ok=True)
|
|
return {
|
|
"status": "rebuild_timeout" if outcome == "timeout" else "rebuild_failed",
|
|
"error": {
|
|
"type": "TimeoutError" if outcome == "timeout" else "WorkerProcessError",
|
|
"message": f"selector candidate rebuild exceeded {timeout_seconds:g} seconds" if outcome == "timeout" else "selector candidate rebuild worker exited without a result",
|
|
},
|
|
}
|
|
|
|
|
|
def _run_comparison(gold_step: Path, rebuilt_step: Path, output: Path, *, timeout_seconds: float) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
|
|
worker = output.with_suffix(".worker.json")
|
|
outcome = _isolated(_compare_worker, (str(gold_step), str(rebuilt_step), str(worker)), worker, timeout_seconds)
|
|
if outcome == "completed":
|
|
comparison = read_json(worker)
|
|
worker.unlink(missing_ok=True)
|
|
return comparison, None
|
|
return None, {
|
|
"type": "TimeoutError" if outcome == "timeout" else "WorkerProcessError",
|
|
"message": f"selector candidate comparison exceeded {timeout_seconds:g} seconds" if outcome == "timeout" else "selector candidate comparison worker exited without a result",
|
|
}
|
|
|
|
|
|
def _outcome_summary(outcome: dict[str, Any], comparison: dict[str, Any] | None) -> dict[str, Any]:
|
|
summary: dict[str, Any] = {
|
|
"rebuild_status": outcome.get("status"),
|
|
"last_executable_prefix": {
|
|
key: value
|
|
for key, value in (outcome.get("last_executable_prefix") or {}).items()
|
|
if key in {"failed_feature_id", "feature_count", "last_feature_id"}
|
|
},
|
|
}
|
|
error = outcome.get("error")
|
|
if isinstance(error, dict):
|
|
summary["error"] = {key: error.get(key) for key in ("type", "message")}
|
|
if comparison is not None:
|
|
summary["comparison"] = {
|
|
"decision": comparison.get("decision"),
|
|
"strict_passed": bool((comparison.get("strict") or {}).get("passed")),
|
|
"rp_passed": bool((comparison.get("rp") or {}).get("passed")),
|
|
}
|
|
return summary
|
|
|
|
|
|
def _strict_passed(comparison: dict[str, Any] | None) -> bool:
|
|
return bool(isinstance(comparison, dict) and (comparison.get("strict") or {}).get("passed"))
|
|
|
|
|
|
_SEARCH_TOKEN = "_selector_search_token"
|
|
|
|
|
|
def _selector_search_token(feature_id: str, location: str) -> str:
|
|
"""Use a structural token while a group expands a public selector list."""
|
|
return f"{feature_id}:{location}"
|
|
|
|
|
|
def _annotate_selector_search_tokens(cdsl: dict[str, Any]) -> int:
|
|
"""Mark copied diagnostic selectors without creating a durable CDSL field."""
|
|
count = 0
|
|
for feature in cdsl.get("features") or []:
|
|
if not isinstance(feature, dict) or not isinstance(feature.get("id"), str):
|
|
continue
|
|
for location, selector in _selector_sites(feature):
|
|
selector[_SEARCH_TOKEN] = _selector_search_token(feature["id"], location)
|
|
count += 1
|
|
return count
|
|
|
|
|
|
def _strip_selector_search_tokens(value: Any) -> int:
|
|
"""Remove experiment-only branch tokens before a CDSL is persisted."""
|
|
removed = 0
|
|
if isinstance(value, dict):
|
|
if _SEARCH_TOKEN in value:
|
|
value.pop(_SEARCH_TOKEN)
|
|
removed += 1
|
|
for child in value.values():
|
|
removed += _strip_selector_search_tokens(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
removed += _strip_selector_search_tokens(child)
|
|
return removed
|
|
|
|
|
|
def _selector_sites(feature: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
|
|
"""Return the public selector positions of one CDSL feature.
|
|
|
|
The locations are deliberately structural rather than runtime IDs. A
|
|
selector record must remain useful when a fresh replay creates different
|
|
topology-record instances.
|
|
"""
|
|
sites: list[tuple[str, dict[str, Any]]] = []
|
|
for index, selector in enumerate(feature.get("selectors") or []):
|
|
if isinstance(selector, dict):
|
|
sites.append((f"selectors[{index}]", selector))
|
|
params = feature.get("params") or {}
|
|
for name in ("end_condition", "reverse_end_condition"):
|
|
condition = params.get(name)
|
|
reference = condition.get("reference") if isinstance(condition, dict) else None
|
|
if isinstance(reference, dict):
|
|
sites.append((f"params.{name}.reference", reference))
|
|
return sites
|
|
|
|
|
|
def _semantic_source_selector(selector: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""Keep the FeatureScript meaning, never a source runtime topology ID."""
|
|
intent = selector.get("selector_intent")
|
|
if not isinstance(intent, dict):
|
|
return None
|
|
result = {
|
|
key: deepcopy(selector[key])
|
|
for key in ("kind", "owner_feature_id", "binding_feature_id", "output_role")
|
|
if selector.get(key) is not None
|
|
}
|
|
result["selector_intent"] = deepcopy(intent)
|
|
return result
|
|
|
|
|
|
def _selector_record_entries(
|
|
sample_id: str,
|
|
provenance_candidate: dict[str, Any],
|
|
compiled_cdsl: dict[str, Any],
|
|
selector_binding: list[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
"""Extract FeatureScript meaning plus a freshly replayable binding plan.
|
|
|
|
A geometry probe has only the selected record, not rejected alternatives.
|
|
We preserve that limitation explicitly. When source and compiled selector
|
|
counts differ (for example, one source query expands to several topology
|
|
records), emit a feature group instead of inventing a one-to-one mapping.
|
|
"""
|
|
source_by_id = {
|
|
str(feature.get("id")): feature
|
|
for feature in provenance_candidate.get("features") or []
|
|
if isinstance(feature, dict) and isinstance(feature.get("id"), str)
|
|
}
|
|
compiled_by_id = {
|
|
str(feature.get("id")): feature
|
|
for feature in compiled_cdsl.get("features") or []
|
|
if isinstance(feature, dict) and isinstance(feature.get("id"), str)
|
|
}
|
|
entries: list[dict[str, Any]] = []
|
|
for binding in selector_binding:
|
|
feature_id = binding.get("feature_id")
|
|
if not isinstance(feature_id, str):
|
|
continue
|
|
source_feature = source_by_id.get(feature_id)
|
|
compiled_feature = compiled_by_id.get(feature_id)
|
|
if source_feature is None or compiled_feature is None:
|
|
continue
|
|
source_sites = [
|
|
(location, semantic)
|
|
for location, selector in _selector_sites(source_feature)
|
|
if (semantic := _semantic_source_selector(selector)) is not None
|
|
]
|
|
if not source_sites:
|
|
continue
|
|
compiled_sites = _selector_sites(compiled_feature)
|
|
resolved = [deepcopy(item) for item in binding.get("resolved") or [] if isinstance(item, dict)]
|
|
feature_context = {
|
|
"sample_id": sample_id,
|
|
"feature_id": feature_id,
|
|
"atomic_id": compiled_feature.get("atomic_id"),
|
|
"prefix_feature_count": binding.get("prefix_feature_count"),
|
|
"verification_scope": "behavioural_strict_replay",
|
|
}
|
|
if len(source_sites) == len(compiled_sites) == len(resolved):
|
|
for index, ((location, source_selector), (_compiled_location, compiled_selector), resolved_record) in enumerate(
|
|
zip(source_sites, compiled_sites, resolved)
|
|
):
|
|
entries.append({
|
|
"schema": "cadfs_to_cdsl.selector_record.v1",
|
|
"entry_id": f"{sample_id}:{feature_id}:{location}",
|
|
**feature_context,
|
|
"source_location": location,
|
|
"source_selector": source_selector,
|
|
"compiled_selector": deepcopy(compiled_selector),
|
|
"resolved_record": resolved_record,
|
|
"mapping": "one_to_one",
|
|
"limitations": [
|
|
"The source dataset has no per-feature Onshape query witness.",
|
|
"The compiled selector is verified only by a fresh local strict replay.",
|
|
"Runtime snapshot IDs are replay evidence, not portable topology identities.",
|
|
],
|
|
})
|
|
continue
|
|
entries.append({
|
|
"schema": "cadfs_to_cdsl.selector_record.v1",
|
|
"entry_id": f"{sample_id}:{feature_id}:feature_group",
|
|
**feature_context,
|
|
"source_selectors": [
|
|
{"location": location, "selector": selector}
|
|
for location, selector in source_sites
|
|
],
|
|
"compiled_selectors": [
|
|
{"location": location, "selector": deepcopy(selector)}
|
|
for location, selector in compiled_sites
|
|
],
|
|
"resolved_records": resolved,
|
|
"mapping": "feature_group",
|
|
"limitations": [
|
|
"The source query expanded or contracted during local binding, so this record does not claim a one-to-one mapping.",
|
|
"The source dataset has no per-feature Onshape query witness.",
|
|
"The compiled selector is verified only by a fresh local strict replay.",
|
|
"Runtime snapshot IDs are replay evidence, not portable topology identities.",
|
|
],
|
|
})
|
|
return entries
|
|
|
|
|
|
def _emit_strict_selector_record(
|
|
sample: Sample,
|
|
directory: Path,
|
|
provenance_candidate: dict[str, Any],
|
|
geometry_probe: dict[str, Any],
|
|
geometry_comparison: dict[str, Any] | None,
|
|
gold_step: Path,
|
|
*,
|
|
rebuild_timeout_seconds: float,
|
|
comparison_timeout_seconds: float,
|
|
query_groups: list[dict[str, Any]] | None = None,
|
|
candidate_replay_kind: str = "geometry_probe",
|
|
) -> dict[str, Any]:
|
|
"""Create and independently verify a compiled CDSL only after strict proof."""
|
|
if not _strict_passed(geometry_comparison):
|
|
return {
|
|
"status": "not_emitted",
|
|
"reason": "geometry_probe_not_strict",
|
|
}
|
|
bound_cdsl = geometry_probe.get("bound_cdsl")
|
|
binding = geometry_probe.get("selector_binding")
|
|
if not isinstance(bound_cdsl, dict) or not isinstance(binding, list):
|
|
return {
|
|
"status": "not_emitted",
|
|
"reason": "geometry_probe_binding_missing",
|
|
}
|
|
|
|
compiled_candidate = deepcopy(bound_cdsl)
|
|
_strip_selector_search_tokens(compiled_candidate)
|
|
compiled_path = directory / "selector_record.compiled.cdsl.json"
|
|
write_json(compiled_path, compiled_candidate)
|
|
compiled = _run_rebuild(
|
|
compiled_path,
|
|
directory / "selector_record.compiled.rebuild.step",
|
|
timeout_seconds=rebuild_timeout_seconds,
|
|
)
|
|
_write_rebuild_artifacts(directory, "selector_record.compiled", compiled)
|
|
comparison = None
|
|
comparison_error = None
|
|
if compiled.get("status") == "rebuilt":
|
|
comparison, comparison_error = _run_comparison(
|
|
gold_step,
|
|
directory / "selector_record.compiled.rebuild.step",
|
|
directory / "selector_record.compiled.comparison.json",
|
|
timeout_seconds=comparison_timeout_seconds,
|
|
)
|
|
if comparison is not None:
|
|
write_json(directory / "selector_record.compiled.comparison.json", comparison)
|
|
|
|
result = {
|
|
"status": "not_emitted",
|
|
"compiled_cdsl": str(compiled_path),
|
|
"compiled_replay": _outcome_summary(compiled, comparison),
|
|
}
|
|
if comparison_error is not None:
|
|
result["compiled_replay"]["comparison_error"] = comparison_error
|
|
if not _strict_passed(comparison):
|
|
result["reason"] = "compiled_cdsl_not_strict"
|
|
return result
|
|
|
|
records = _selector_record_entries(sample.sample_id, provenance_candidate, compiled_candidate, binding)
|
|
if not records:
|
|
result["reason"] = "no_featurescript_selector_binding"
|
|
return result
|
|
record_path = directory / "selector-record.json"
|
|
verification = {
|
|
"compiled_cdsl_strict": True,
|
|
"compiled_cdsl": str(compiled_path),
|
|
}
|
|
if candidate_replay_kind == "geometry_probe":
|
|
verification["geometry_probe_strict"] = True
|
|
else:
|
|
verification["selector_search_branch_strict"] = True
|
|
record_payload: dict[str, Any] = {
|
|
"schema": "cadfs_to_cdsl.selector_record_set.v1",
|
|
"sample_id": sample.sample_id,
|
|
"source": {
|
|
"featurescript": sample.files.get("featurescript"),
|
|
"step": sample.files.get("step"),
|
|
},
|
|
"verification": verification,
|
|
"records": records,
|
|
}
|
|
if query_groups:
|
|
# These are the query-to-record *sets* selected by the offline search.
|
|
# A FeatureScript query is not necessarily a single OCC record.
|
|
record_payload["query_groups"] = deepcopy(query_groups)
|
|
write_json(record_path, record_payload)
|
|
result.update({
|
|
"status": "strict_replayed",
|
|
"record_count": len(records),
|
|
"record_path": str(record_path),
|
|
})
|
|
return result
|
|
|
|
|
|
def _searchable_selector_sites(feature: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
|
|
"""Return direct selector roots that can be replaced as a search branch.
|
|
|
|
The experimental search does not flatten an intersection query into an
|
|
unrelated member. It can search direct roots and extent references; other
|
|
query trees remain a bounded unsupported search outcome.
|
|
"""
|
|
sites = _selector_sites(feature)
|
|
return [
|
|
(location, selector)
|
|
for location, selector in sites
|
|
if not isinstance(selector.get("intersection_of"), list)
|
|
]
|
|
|
|
|
|
def _replace_selector_group(
|
|
cdsl: dict[str, Any],
|
|
feature_id: str,
|
|
token: str,
|
|
selectors: list[dict[str, Any]],
|
|
) -> None:
|
|
"""Expand one selector root to an explicit branch-local record set."""
|
|
feature = next(
|
|
(item for item in cdsl.get("features") or [] if isinstance(item, dict) and item.get("id") == feature_id),
|
|
None,
|
|
)
|
|
if feature is None:
|
|
raise ValueError(f"search branch cannot find feature {feature_id}")
|
|
roots = feature.get("selectors") or []
|
|
index = next(
|
|
(
|
|
item_index
|
|
for item_index, selector in enumerate(roots)
|
|
if isinstance(selector, dict) and selector.get(_SEARCH_TOKEN) == token
|
|
),
|
|
None,
|
|
)
|
|
if index is not None:
|
|
roots[index:index + 1] = [deepcopy(selector) for selector in selectors]
|
|
feature["selectors"] = roots
|
|
return
|
|
for name in ("end_condition", "reverse_end_condition"):
|
|
condition = (feature.get("params") or {}).get(name)
|
|
reference = condition.get("reference") if isinstance(condition, dict) else None
|
|
if isinstance(reference, dict) and reference.get(_SEARCH_TOKEN) == token:
|
|
if len(selectors) != 1:
|
|
raise ValueError("search group cannot expand an extent reference")
|
|
condition["reference"] = deepcopy(selectors[0])
|
|
return
|
|
raise ValueError(f"search branch selector token is unavailable: {feature_id}:{token}")
|
|
|
|
|
|
def _forced_candidate_selectors(placeholder: dict[str, Any], candidates: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Turn a query's runtime record set into explicit branch-local roots."""
|
|
from .selector_binding import _bound_selector
|
|
|
|
selectors: list[dict[str, Any]] = []
|
|
for candidate in candidates:
|
|
selector, _bound = _bound_selector(placeholder, [candidate])
|
|
if not isinstance(selector, dict):
|
|
raise ValueError("search branch produced an invalid explicit selector")
|
|
if placeholder.get(_SEARCH_TOKEN) is not None:
|
|
selector[_SEARCH_TOKEN] = placeholder[_SEARCH_TOKEN]
|
|
selectors.append(selector)
|
|
if not selectors:
|
|
raise ValueError("search group has no runtime records")
|
|
return selectors
|
|
|
|
|
|
def _resolve_search_selector(placeholder: dict[str, Any], *, registry: Any, active_body_id: str | None) -> Any:
|
|
"""Mirror the legacy diagnostic binding policy without changing it."""
|
|
from .selector_binding import _runtime_selector
|
|
|
|
runtime_selector = _runtime_selector(placeholder)
|
|
resolution = registry.resolve(runtime_selector, active_body_id=active_body_id)
|
|
intent = placeholder.get("selector_intent")
|
|
if (
|
|
resolution.status != "resolved"
|
|
and not isinstance(intent, dict)
|
|
and not runtime_selector.get("owner_match_required")
|
|
and not runtime_selector.get("output_role")
|
|
):
|
|
fallback = dict(runtime_selector)
|
|
fallback.pop("owner_feature_id", None)
|
|
resolution = registry.resolve(fallback, active_body_id=active_body_id)
|
|
return resolution
|
|
|
|
|
|
def _search_candidate_records(
|
|
resolution: Any,
|
|
*,
|
|
maximum: int,
|
|
minimum_score: float = 0.8,
|
|
) -> tuple[list[dict[str, Any]], int]:
|
|
"""Keep only candidates the normal resolver considers geometrically valid."""
|
|
candidates = [
|
|
item for item in resolution.candidates
|
|
if isinstance(item, dict)
|
|
and item.get("record_id")
|
|
and float(item.get("score") or 0.0) >= minimum_score
|
|
]
|
|
candidates.sort(key=lambda item: (-float(item.get("score") or 0.0), str(item.get("record_id"))))
|
|
return [deepcopy(item) for item in candidates[:maximum]], len(candidates)
|
|
|
|
|
|
def _vector(value: Any) -> list[float] | None:
|
|
if not isinstance(value, (list, tuple)) or len(value) != 3:
|
|
return None
|
|
try:
|
|
result = [float(component) for component in value]
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return result if all(math.isfinite(component) for component in result) else None
|
|
|
|
|
|
def _sub(left: list[float], right: list[float]) -> list[float]:
|
|
return [left[index] - right[index] for index in range(3)]
|
|
|
|
|
|
def _dot(left: list[float], right: list[float]) -> float:
|
|
return sum(left[index] * right[index] for index in range(3))
|
|
|
|
|
|
def _cross(left: list[float], right: list[float]) -> list[float]:
|
|
return [
|
|
left[1] * right[2] - left[2] * right[1],
|
|
left[2] * right[0] - left[0] * right[2],
|
|
left[0] * right[1] - left[1] * right[0],
|
|
]
|
|
|
|
|
|
def _norm(value: list[float]) -> float:
|
|
return math.sqrt(_dot(value, value))
|
|
|
|
|
|
def _unit(value: list[float] | None) -> list[float] | None:
|
|
if value is None or len(value) != 3:
|
|
return None
|
|
length = _norm(value)
|
|
if length <= 1e-9:
|
|
return None
|
|
return [component / length for component in value]
|
|
|
|
|
|
def _distance(left: list[float], right: list[float]) -> float:
|
|
return _norm(_sub(left, right))
|
|
|
|
|
|
def _score_distance(distance_mm: float, *, tolerance_mm: float = 0.05) -> float:
|
|
return max(0.0, 1.0 - distance_mm / tolerance_mm)
|
|
|
|
|
|
def _source_selector_map(provenance_candidate: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
"""Index source selectors by an experiment-only structural token."""
|
|
selectors: dict[str, dict[str, Any]] = {}
|
|
for feature in provenance_candidate.get("features") or []:
|
|
if not isinstance(feature, dict) or not isinstance(feature.get("id"), str):
|
|
continue
|
|
for location, selector in _selector_sites(feature):
|
|
if isinstance(selector.get("selector_intent"), dict):
|
|
selectors[_selector_search_token(feature["id"], location)] = deepcopy(selector)
|
|
return selectors
|
|
|
|
|
|
def _source_feature(provenance_candidate: dict[str, Any], feature_id: Any) -> dict[str, Any] | None:
|
|
if not isinstance(feature_id, str):
|
|
return None
|
|
return next(
|
|
(
|
|
feature
|
|
for feature in provenance_candidate.get("features") or []
|
|
if isinstance(feature, dict) and feature.get("id") == feature_id
|
|
),
|
|
None,
|
|
)
|
|
|
|
|
|
def _source_sketch(provenance_candidate: dict[str, Any], producer: dict[str, Any], source_sketch_id: str | None) -> dict[str, Any] | None:
|
|
sketches = (provenance_candidate.get("geometry") or {}).get("sketches") or []
|
|
sketch_id = producer.get("sketch_id")
|
|
if isinstance(sketch_id, str):
|
|
sketch = next((item for item in sketches if isinstance(item, dict) and item.get("id") == sketch_id), None)
|
|
if isinstance(sketch, dict):
|
|
return sketch
|
|
if isinstance(source_sketch_id, str):
|
|
sketch = next(
|
|
(
|
|
item
|
|
for item in sketches
|
|
if isinstance(item, dict) and item.get("source_sketch_id") == source_sketch_id
|
|
),
|
|
None,
|
|
)
|
|
if isinstance(sketch, dict):
|
|
return sketch
|
|
return None
|
|
|
|
|
|
def _source_curve(sketch: dict[str, Any], entity_id: str) -> dict[str, Any] | None:
|
|
profile = sketch.get("profile") or {}
|
|
for entry in profile.get("source_entities") or []:
|
|
if isinstance(entry, dict) and entry.get("id") == entity_id and isinstance(entry.get("curve"), dict):
|
|
return entry["curve"]
|
|
for contour in profile.get("contours") or []:
|
|
if not isinstance(contour, dict):
|
|
continue
|
|
for segment in contour.get("segments") or []:
|
|
if isinstance(segment, dict) and segment.get("source_entity_id") == entity_id:
|
|
return segment
|
|
for segment in profile.get("construction") or []:
|
|
if isinstance(segment, dict) and segment.get("source_entity_id") == entity_id:
|
|
return segment
|
|
return None
|
|
|
|
|
|
def _producer_frame(
|
|
provenance_candidate: dict[str, Any],
|
|
source_selector: dict[str, Any],
|
|
) -> tuple[dict[str, Any], dict[str, Any], list[float], float] | None:
|
|
"""Recover only the direct sketch/extrude frame needed for candidate rank."""
|
|
intent = source_selector.get("selector_intent") or {}
|
|
source_entity = intent.get("source_entity") if isinstance(intent.get("source_entity"), dict) else None
|
|
source_entities = intent.get("source_entities") or []
|
|
source_sketch_id = (
|
|
source_entity.get("sketch_id")
|
|
if isinstance(source_entity, dict)
|
|
else next(
|
|
(
|
|
item.get("sketch_id")
|
|
for item in source_entities
|
|
if isinstance(item, dict) and isinstance(item.get("sketch_id"), str)
|
|
),
|
|
None,
|
|
)
|
|
)
|
|
producer = _source_feature(provenance_candidate, source_selector.get("owner_feature_id"))
|
|
if producer is None or not str(producer.get("atomic_id") or "").startswith("extrude_"):
|
|
return None
|
|
sketch = _source_sketch(provenance_candidate, producer, source_sketch_id)
|
|
if sketch is None:
|
|
return None
|
|
plane = sketch.get("workplane") or {}
|
|
origin = _vector(plane.get("origin_mm"))
|
|
normal = _unit(_vector(plane.get("normal")) or [])
|
|
if origin is None or normal is None:
|
|
return None
|
|
params = producer.get("params") or {}
|
|
try:
|
|
span = abs(float(params["distance_mm"]))
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
if span <= 1e-9:
|
|
return None
|
|
if bool(params.get("reverse") or params.get("opposite_direction")):
|
|
normal = [-component for component in normal]
|
|
return producer, sketch, normal, span
|
|
|
|
|
|
def _world_point(sketch: dict[str, Any], point: Any) -> list[float] | None:
|
|
local = point if isinstance(point, (list, tuple)) and len(point) in {2, 3} else None
|
|
if local is None:
|
|
return None
|
|
try:
|
|
coordinates = [float(value) for value in local]
|
|
except (TypeError, ValueError):
|
|
return None
|
|
plane = sketch.get("workplane") or {}
|
|
origin = _vector(plane.get("origin_mm"))
|
|
x_axis = _unit(_vector(plane.get("x_dir")) or [])
|
|
normal = _unit(_vector(plane.get("normal")) or [])
|
|
if origin is None or x_axis is None or normal is None:
|
|
return None
|
|
y_axis = _unit(_cross(normal, x_axis))
|
|
if y_axis is None:
|
|
return None
|
|
z = coordinates[2] if len(coordinates) == 3 else 0.0
|
|
return [
|
|
origin[index] + x_axis[index] * coordinates[0] + y_axis[index] * coordinates[1] + normal[index] * z
|
|
for index in range(3)
|
|
]
|
|
|
|
|
|
def _source_entity_ids(source_selector: dict[str, Any]) -> list[str]:
|
|
intent = source_selector.get("selector_intent") or {}
|
|
entity = intent.get("source_entity")
|
|
if isinstance(entity, dict) and isinstance(entity.get("entity_id"), str):
|
|
return [entity["entity_id"]]
|
|
return sorted({
|
|
item["entity_id"]
|
|
for item in intent.get("source_entities") or []
|
|
if isinstance(item, dict) and isinstance(item.get("entity_id"), str)
|
|
})
|
|
|
|
|
|
def _cap_role_hint(source_selector: dict[str, Any]) -> str | None:
|
|
intent = source_selector.get("selector_intent") or {}
|
|
role = source_selector.get("output_role") or intent.get("output_role")
|
|
if isinstance(role, str) and role in {"extrude.start", "extrude.end"}:
|
|
return role.rsplit(".", 1)[1]
|
|
ast = ((intent.get("source_query") or {}).get("ast") or {})
|
|
args = ast.get("args") if isinstance(ast, dict) else None
|
|
options = args[3] if isinstance(args, list) and len(args) >= 4 and isinstance(args[3], dict) else {}
|
|
is_start = options.get("isStart")
|
|
if is_start in {True, "true"}:
|
|
return "start"
|
|
if is_start in {False, "false"}:
|
|
return "end"
|
|
return None
|
|
|
|
|
|
def _query_group_descriptor(token: str, source_selector: dict[str, Any]) -> dict[str, Any]:
|
|
intent = source_selector.get("selector_intent") or {}
|
|
policy = intent.get("derivation_policy") or {}
|
|
return {
|
|
"token": token,
|
|
"feature_id": token.split(":", 1)[0],
|
|
"source_location": token.split(":", 1)[1] if ":" in token else token,
|
|
"kind": source_selector.get("kind"),
|
|
"owner_feature_id": source_selector.get("owner_feature_id"),
|
|
"query_family": intent.get("query_family"),
|
|
"multiplicity": policy.get("multiplicity", "one"),
|
|
"source_entity_ids": _source_entity_ids(source_selector),
|
|
"source_make_query": deepcopy((intent.get("source_query") or {}).get("ast")),
|
|
}
|
|
|
|
|
|
def _rank_cap_face(
|
|
provenance_candidate: dict[str, Any],
|
|
source_selector: dict[str, Any],
|
|
candidate: dict[str, Any],
|
|
) -> tuple[float, dict[str, Any]] | None:
|
|
frame = _producer_frame(provenance_candidate, source_selector)
|
|
role = _cap_role_hint(source_selector)
|
|
if frame is None or role is None:
|
|
return None
|
|
_producer, sketch, direction, span = frame
|
|
origin = _vector((sketch.get("workplane") or {}).get("origin_mm"))
|
|
geometry = candidate.get("geometry") or {}
|
|
actual_normal = _unit(_vector(geometry.get("plane_normal") or geometry.get("normal")) or [])
|
|
if origin is None or actual_normal is None or geometry.get("surface_type") != "plane":
|
|
return None
|
|
point = origin if role == "start" else [origin[index] + direction[index] * span for index in range(3)]
|
|
expected_offset = _dot(direction, point)
|
|
actual_offset = geometry.get("plane_offset_mm")
|
|
try:
|
|
actual_offset = float(actual_offset)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
alignment = abs(_dot(direction, actual_normal))
|
|
signed_offset = actual_offset if _dot(direction, actual_normal) >= 0 else -actual_offset
|
|
offset_error = abs(expected_offset - signed_offset)
|
|
return (alignment + _score_distance(offset_error)) / 2.0, {
|
|
"method": "extrude_cap_plane",
|
|
"cap_role": role,
|
|
"normal_alignment": round(alignment, 6),
|
|
"plane_offset_error_mm": round(offset_error, 9),
|
|
}
|
|
|
|
|
|
def _rank_swept_face(
|
|
provenance_candidate: dict[str, Any],
|
|
source_selector: dict[str, Any],
|
|
candidate: dict[str, Any],
|
|
) -> tuple[float, dict[str, Any]] | None:
|
|
frame = _producer_frame(provenance_candidate, source_selector)
|
|
entity_ids = _source_entity_ids(source_selector)
|
|
if frame is None or len(entity_ids) != 1:
|
|
return None
|
|
_producer, sketch, direction, _span = frame
|
|
curve = _source_curve(sketch, entity_ids[0])
|
|
geometry = candidate.get("geometry") or {}
|
|
if not isinstance(curve, dict):
|
|
return None
|
|
if curve.get("type") == "line":
|
|
start, end = _world_point(sketch, curve.get("start")), _world_point(sketch, curve.get("end"))
|
|
actual_normal = _unit(_vector(geometry.get("plane_normal") or geometry.get("normal")) or [])
|
|
if start is None or end is None or actual_normal is None or geometry.get("surface_type") != "plane":
|
|
return None
|
|
expected_normal = _unit(_cross(_sub(end, start), direction))
|
|
if expected_normal is None:
|
|
return None
|
|
try:
|
|
actual_offset = float(geometry["plane_offset_mm"])
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
alignment = abs(_dot(expected_normal, actual_normal))
|
|
expected_offset = _dot(expected_normal, start)
|
|
signed_offset = actual_offset if _dot(expected_normal, actual_normal) >= 0 else -actual_offset
|
|
offset_error = abs(expected_offset - signed_offset)
|
|
return (alignment + _score_distance(offset_error)) / 2.0, {
|
|
"method": "source_line_supporting_plane",
|
|
"normal_alignment": round(alignment, 6),
|
|
"plane_offset_error_mm": round(offset_error, 9),
|
|
}
|
|
if curve.get("type") not in {"arc", "circle"}:
|
|
return None
|
|
center = _world_point(sketch, curve.get("center"))
|
|
actual_origin = _vector(geometry.get("axis_origin_mm"))
|
|
actual_direction = _unit(_vector(geometry.get("axis_direction")) or [])
|
|
try:
|
|
radius_error = abs(float(curve["radius_mm"]) - float(geometry["radius_mm"]))
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
if center is None or actual_origin is None or actual_direction is None or geometry.get("surface_type") != "cylinder":
|
|
return None
|
|
alignment = abs(_dot(direction, actual_direction))
|
|
radial_offset = _norm(_cross(_sub(center, actual_origin), actual_direction))
|
|
return (
|
|
alignment + _score_distance(radial_offset) + _score_distance(radius_error)
|
|
) / 3.0, {
|
|
"method": "source_arc_cylinder",
|
|
"axis_alignment": round(alignment, 6),
|
|
"axis_offset_mm": round(radial_offset, 9),
|
|
"radius_error_mm": round(radius_error, 9),
|
|
}
|
|
|
|
|
|
def _point2(value: Any) -> tuple[float, float] | None:
|
|
if not isinstance(value, (list, tuple)) or len(value) < 2:
|
|
return None
|
|
try:
|
|
point = float(value[0]), float(value[1])
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return point if all(math.isfinite(component) for component in point) else None
|
|
|
|
|
|
def _curve_contains_point(curve: dict[str, Any], point: tuple[float, float], *, tolerance_mm: float = 0.05) -> bool:
|
|
curve_type = curve.get("type")
|
|
if curve_type == "line":
|
|
start, end = _point2(curve.get("start")), _point2(curve.get("end"))
|
|
if start is None or end is None:
|
|
return False
|
|
direction = end[0] - start[0], end[1] - start[1]
|
|
length = math.hypot(*direction)
|
|
if length <= 1e-9:
|
|
return False
|
|
offset = point[0] - start[0], point[1] - start[1]
|
|
distance = abs(direction[0] * offset[1] - direction[1] * offset[0]) / length
|
|
parameter = (direction[0] * offset[0] + direction[1] * offset[1]) / (length * length)
|
|
return distance <= tolerance_mm and -tolerance_mm / length <= parameter <= 1.0 + tolerance_mm / length
|
|
center = _point2(curve.get("center"))
|
|
try:
|
|
radius = float(curve["radius_mm"])
|
|
except (KeyError, TypeError, ValueError):
|
|
return False
|
|
if center is None or radius <= 1e-9 or abs(math.dist(center, point) - radius) > tolerance_mm:
|
|
return False
|
|
if curve_type == "circle":
|
|
return True
|
|
if curve_type != "arc":
|
|
return False
|
|
start, end = _point2(curve.get("start")), _point2(curve.get("end"))
|
|
if start is None or end is None:
|
|
return False
|
|
angle = math.atan2(point[1] - center[1], point[0] - center[0])
|
|
start_angle = math.atan2(start[1] - center[1], start[0] - center[0])
|
|
end_angle = math.atan2(end[1] - center[1], end[0] - center[0])
|
|
full_turn = 2.0 * math.pi
|
|
if bool(curve.get("clockwise")):
|
|
travelled = (start_angle - angle) % full_turn
|
|
total = (start_angle - end_angle) % full_turn
|
|
else:
|
|
travelled = (angle - start_angle) % full_turn
|
|
total = (end_angle - start_angle) % full_turn
|
|
angular_tolerance = tolerance_mm / radius
|
|
return travelled <= total + angular_tolerance
|
|
|
|
|
|
def _curve_intersections_2d(left: dict[str, Any], right: dict[str, Any]) -> list[tuple[float, float]]:
|
|
"""Intersect line/arc/circle supports, retaining only source curve spans."""
|
|
left_type, right_type = left.get("type"), right.get("type")
|
|
points: list[tuple[float, float]] = []
|
|
if left_type == "line" and right_type == "line":
|
|
left_start, left_end = _point2(left.get("start")), _point2(left.get("end"))
|
|
right_start, right_end = _point2(right.get("start")), _point2(right.get("end"))
|
|
if None not in (left_start, left_end, right_start, right_end):
|
|
left_dx, left_dy = left_end[0] - left_start[0], left_end[1] - left_start[1]
|
|
right_dx, right_dy = right_end[0] - right_start[0], right_end[1] - right_start[1]
|
|
determinant = left_dx * right_dy - left_dy * right_dx
|
|
if abs(determinant) > 1e-9:
|
|
offset_x, offset_y = right_start[0] - left_start[0], right_start[1] - left_start[1]
|
|
parameter = (offset_x * right_dy - offset_y * right_dx) / determinant
|
|
points.append((left_start[0] + parameter * left_dx, left_start[1] + parameter * left_dy))
|
|
else:
|
|
line, circular = (left, right) if left_type == "line" else (right, left)
|
|
if line.get("type") == "line" and circular.get("type") in {"arc", "circle"}:
|
|
start, end, center = _point2(line.get("start")), _point2(line.get("end")), _point2(circular.get("center"))
|
|
try:
|
|
radius = float(circular["radius_mm"])
|
|
except (KeyError, TypeError, ValueError):
|
|
radius = 0.0
|
|
if start is not None and end is not None and center is not None and radius > 1e-9:
|
|
dx, dy = end[0] - start[0], end[1] - start[1]
|
|
ox, oy = start[0] - center[0], start[1] - center[1]
|
|
quadratic_a = dx * dx + dy * dy
|
|
quadratic_b = 2.0 * (dx * ox + dy * oy)
|
|
quadratic_c = ox * ox + oy * oy - radius * radius
|
|
discriminant = quadratic_b * quadratic_b - 4.0 * quadratic_a * quadratic_c
|
|
if quadratic_a > 1e-12 and discriminant >= -1e-9:
|
|
root = math.sqrt(max(0.0, discriminant))
|
|
for parameter in ((-quadratic_b - root) / (2.0 * quadratic_a), (-quadratic_b + root) / (2.0 * quadratic_a)):
|
|
points.append((start[0] + parameter * dx, start[1] + parameter * dy))
|
|
elif left_type in {"arc", "circle"} and right_type in {"arc", "circle"}:
|
|
left_center, right_center = _point2(left.get("center")), _point2(right.get("center"))
|
|
try:
|
|
left_radius, right_radius = float(left["radius_mm"]), float(right["radius_mm"])
|
|
except (KeyError, TypeError, ValueError):
|
|
left_radius = right_radius = 0.0
|
|
if left_center is not None and right_center is not None and left_radius > 1e-9 and right_radius > 1e-9:
|
|
dx, dy = right_center[0] - left_center[0], right_center[1] - left_center[1]
|
|
center_distance = math.hypot(dx, dy)
|
|
if 1e-9 < center_distance <= left_radius + right_radius + 0.05:
|
|
along = (left_radius * left_radius - right_radius * right_radius + center_distance * center_distance) / (2.0 * center_distance)
|
|
height_squared = left_radius * left_radius - along * along
|
|
if height_squared >= -1e-9:
|
|
height = math.sqrt(max(0.0, height_squared))
|
|
base_x, base_y = left_center[0] + along * dx / center_distance, left_center[1] + along * dy / center_distance
|
|
offset_x, offset_y = -dy * height / center_distance, dx * height / center_distance
|
|
points.extend(((base_x + offset_x, base_y + offset_y), (base_x - offset_x, base_y - offset_y)))
|
|
valid: list[tuple[float, float]] = []
|
|
for point in points:
|
|
if not _curve_contains_point(left, point) or not _curve_contains_point(right, point):
|
|
continue
|
|
if not any(math.dist(point, existing) <= 1e-6 for existing in valid):
|
|
valid.append(point)
|
|
return valid
|
|
|
|
|
|
def _source_curve_intersection(sketch: dict[str, Any], entity_ids: list[str]) -> list[float] | None:
|
|
if len(entity_ids) != 2:
|
|
return None
|
|
curves = [_source_curve(sketch, entity_id) for entity_id in entity_ids]
|
|
if any(not isinstance(curve, dict) for curve in curves):
|
|
return None
|
|
intersections = _curve_intersections_2d(curves[0], curves[1])
|
|
if len(intersections) != 1:
|
|
return None
|
|
return _world_point(sketch, intersections[0])
|
|
|
|
|
|
def _rank_swept_edge(
|
|
provenance_candidate: dict[str, Any],
|
|
source_selector: dict[str, Any],
|
|
candidate: dict[str, Any],
|
|
) -> tuple[float, dict[str, Any]] | None:
|
|
frame = _producer_frame(provenance_candidate, source_selector)
|
|
entity_ids = _source_entity_ids(source_selector)
|
|
if frame is None:
|
|
return None
|
|
_producer, sketch, direction, _span = frame
|
|
source_point = _source_curve_intersection(sketch, entity_ids)
|
|
geometry = candidate.get("geometry") or {}
|
|
start, end = _vector(geometry.get("start_mm")), _vector(geometry.get("end_mm"))
|
|
if source_point is None or start is None or end is None or geometry.get("curve_type") != "line":
|
|
return None
|
|
edge_direction = _unit(_sub(end, start))
|
|
if edge_direction is None:
|
|
return None
|
|
alignment = abs(_dot(direction, edge_direction))
|
|
supporting_line_error = _norm(_cross(_sub(source_point, start), direction))
|
|
return (alignment + _score_distance(supporting_line_error)) / 2.0, {
|
|
"method": "source_pair_extrusion_vertex",
|
|
"axis_alignment": round(alignment, 6),
|
|
"source_vertex_offset_mm": round(supporting_line_error, 9),
|
|
}
|
|
|
|
|
|
def _query_candidate_groups(
|
|
provenance_candidate: dict[str, Any],
|
|
token: str,
|
|
source_selector: dict[str, Any] | None,
|
|
resolution: Any,
|
|
*,
|
|
maximum: int,
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]], int]:
|
|
"""Rank branch alternatives as FeatureScript query result sets.
|
|
|
|
This is deliberately an offline geometric heuristic. It gets one chance
|
|
to reduce a query to record *sets*, then final STEP strict replay decides
|
|
whether that set is useful. It never adds a geometry relation to the
|
|
production selector graph.
|
|
"""
|
|
normal_candidates, normal_count = _search_candidate_records(resolution, maximum=maximum)
|
|
if not isinstance(source_selector, dict):
|
|
descriptor = {"token": token, "ranking_status": "source_query_missing"}
|
|
return descriptor, [
|
|
{"records": [candidate], "score": candidate["score"], "ranking": {"method": "resolver_candidate"}}
|
|
for candidate in normal_candidates
|
|
], normal_count
|
|
descriptor = _query_group_descriptor(token, source_selector)
|
|
family = descriptor.get("query_family")
|
|
ranker = {
|
|
"CAP_FACE": _rank_cap_face,
|
|
"SWEPT_FACE": _rank_swept_face,
|
|
"SWEPT_EDGE": _rank_swept_edge,
|
|
}.get(family)
|
|
if ranker is None:
|
|
descriptor["ranking_status"] = "query_family_not_ranked"
|
|
return descriptor, [
|
|
{"records": [candidate], "score": candidate["score"], "ranking": {"method": "resolver_candidate"}}
|
|
for candidate in normal_candidates
|
|
], normal_count
|
|
ranked: list[tuple[float, dict[str, Any], dict[str, Any]]] = []
|
|
for candidate in resolution.candidates:
|
|
if not isinstance(candidate, dict) or not candidate.get("record_id") or float(candidate.get("score") or 0.0) < 0.8:
|
|
continue
|
|
result = ranker(provenance_candidate, source_selector, candidate)
|
|
if result is None:
|
|
continue
|
|
score, reason = result
|
|
if score >= 0.98:
|
|
ranked.append((score, deepcopy(candidate), reason))
|
|
ranked.sort(key=lambda item: (-item[0], str(item[1].get("record_id"))))
|
|
if not ranked:
|
|
descriptor["ranking_status"] = "source_geometry_unavailable"
|
|
return descriptor, [
|
|
{"records": [candidate], "score": candidate["score"], "ranking": {"method": "resolver_candidate"}}
|
|
for candidate in normal_candidates
|
|
], normal_count
|
|
descriptor["ranking_status"] = "source_geometry_ranked"
|
|
descriptor["ranked_record_count"] = len(ranked)
|
|
if descriptor.get("multiplicity") == "all_fragments":
|
|
records = [candidate for _score, candidate, _reason in ranked]
|
|
descriptor["selected_set_cardinality"] = len(records)
|
|
return descriptor, [{
|
|
"records": records,
|
|
"score": round(sum(score for score, _candidate, _reason in ranked) / len(ranked), 6),
|
|
"ranking": {"method": "all_fragments_source_geometry", "records": [reason for _score, _candidate, reason in ranked]},
|
|
}], 1
|
|
groups = [
|
|
{"records": [candidate], "score": round(score, 6), "ranking": reason}
|
|
for score, candidate, reason in ranked[:maximum]
|
|
]
|
|
return descriptor, groups, len(ranked)
|
|
|
|
|
|
def _search_replay_worker(
|
|
candidate: str,
|
|
provenance: str,
|
|
step: str,
|
|
result: str,
|
|
maximum_candidates: int,
|
|
) -> None:
|
|
"""Replay one branch until it builds or exposes its next selector decision."""
|
|
from engine.cdsl_engine.runtime import finalize_cdsl_execution, prepare_cdsl_execution
|
|
from .selector_binding import _bound_selector, _selector_key
|
|
|
|
bound = read_json(Path(candidate))
|
|
provenance_candidate = read_json(Path(provenance))
|
|
source_selectors = _source_selector_map(provenance_candidate)
|
|
execution = prepare_cdsl_execution(bound)
|
|
if not execution.analysis.runtime_eligible:
|
|
first = next((item for item in execution.analysis.feature_results if not item.executable), None)
|
|
code = first.blockers[0].code if first and first.blockers else "runtime_ineligible"
|
|
write_json(Path(result), {"status": "runtime_ineligible", "reason": code})
|
|
return
|
|
bound_features = {
|
|
str(feature.get("id") or ""): feature
|
|
for feature in bound.get("features") or []
|
|
if isinstance(feature, dict)
|
|
}
|
|
body_id_for_feature: dict[str, str | None] = {"__current__": None}
|
|
evidence: list[dict[str, Any]] = []
|
|
for index, node in enumerate(execution.analysis.plan):
|
|
feature = node.source_feature
|
|
sites = _searchable_selector_sites(feature)
|
|
if len(sites) != len(_selector_sites(feature)):
|
|
write_json(Path(result), {
|
|
"status": "search_unsupported",
|
|
"feature_id": node.feature_id,
|
|
"reason": "nested_selector_tree",
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
})
|
|
return
|
|
resolved: list[dict[str, Any]] = []
|
|
for location, placeholder in sites:
|
|
binding_feature_id = placeholder.get("binding_feature_id")
|
|
if binding_feature_id is None:
|
|
active_body_id = body_id_for_feature["__current__"]
|
|
elif isinstance(binding_feature_id, str) and binding_feature_id in body_id_for_feature:
|
|
active_body_id = body_id_for_feature[binding_feature_id]
|
|
else:
|
|
write_json(Path(result), {
|
|
"status": "search_failed",
|
|
"feature_id": node.feature_id,
|
|
"reason": "selector_binding_feature_missing_or_forward",
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
})
|
|
return
|
|
resolution = _resolve_search_selector(
|
|
placeholder,
|
|
registry=execution.session.topology,
|
|
active_body_id=active_body_id,
|
|
)
|
|
if resolution.status != "resolved":
|
|
token = placeholder.get(_SEARCH_TOKEN)
|
|
descriptor, candidate_groups, group_count = _query_candidate_groups(
|
|
provenance_candidate,
|
|
str(token or f"{node.feature_id}:{location}"),
|
|
source_selectors.get(str(token)),
|
|
resolution,
|
|
maximum=maximum_candidates,
|
|
)
|
|
if candidate_groups:
|
|
write_json(Path(result), {
|
|
"status": "branchable_query_group",
|
|
"feature_id": node.feature_id,
|
|
"source_location": location,
|
|
"selector_token": token,
|
|
"resolution_status": resolution.status,
|
|
"diagnostic": resolution.diagnostic.as_dict() if resolution.diagnostic is not None else None,
|
|
"query_group": descriptor,
|
|
"candidate_groups": candidate_groups,
|
|
"candidate_group_count": group_count,
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
})
|
|
return
|
|
code = resolution.diagnostic.code if resolution.diagnostic is not None else f"selector_{resolution.status}"
|
|
write_json(Path(result), {
|
|
"status": "search_failed",
|
|
"feature_id": node.feature_id,
|
|
"reason": code,
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
})
|
|
return
|
|
selected = list(resolution.records or ((resolution.record,) if resolution.record is not None else ()))
|
|
if not selected:
|
|
write_json(Path(result), {
|
|
"status": "search_failed",
|
|
"feature_id": node.feature_id,
|
|
"reason": "selector_not_found",
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
})
|
|
return
|
|
public_records = [record.public_dict() for record in selected]
|
|
selector, bound_selectors = _bound_selector(placeholder, public_records)
|
|
placeholder.clear()
|
|
placeholder.update(selector)
|
|
resolved.extend(bound_selectors)
|
|
feature["selectors"] = list({
|
|
_selector_key(selector): selector
|
|
for selector in feature.get("selectors") or []
|
|
}.values())
|
|
if feature.get("atomic_id") == "pattern_mirror":
|
|
planes = [selector for selector in feature["selectors"] if selector.get("kind") == "plane"]
|
|
if len(planes) != 1:
|
|
write_json(Path(result), {
|
|
"status": "search_failed",
|
|
"feature_id": node.feature_id,
|
|
"reason": "mirror_plane_binding_not_unique",
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
})
|
|
return
|
|
feature.setdefault("params", {})["mirror_plane"] = planes[0]
|
|
public_feature = bound_features[node.feature_id]
|
|
public_feature["selectors"] = deepcopy(feature.get("selectors") or [])
|
|
public_feature["params"] = deepcopy(feature.get("params") or {})
|
|
evidence.append({
|
|
"feature_id": node.feature_id,
|
|
"prefix_feature_count": index,
|
|
"selectors": public_feature["selectors"],
|
|
"resolved": resolved,
|
|
})
|
|
try:
|
|
execution.execute_next(strict=True)
|
|
except Exception as error:
|
|
write_json(Path(result), {
|
|
"status": "search_failed",
|
|
"feature_id": node.feature_id,
|
|
"reason": str(error),
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
})
|
|
return
|
|
body_id_for_feature[node.feature_id] = execution.session.body_id
|
|
body_id_for_feature["__current__"] = execution.session.body_id
|
|
write_json(Path(result), {
|
|
"status": "rebuilt",
|
|
"bound_cdsl": bound,
|
|
"selector_binding": evidence,
|
|
"result": finalize_cdsl_execution(execution, Path(step)),
|
|
})
|
|
|
|
|
|
def _run_search_replay(
|
|
candidate_path: Path,
|
|
provenance_path: Path,
|
|
step: Path,
|
|
*,
|
|
timeout_seconds: float,
|
|
maximum_candidates: int,
|
|
) -> dict[str, Any]:
|
|
worker = step.with_suffix(".worker.json")
|
|
outcome = _isolated(
|
|
_search_replay_worker,
|
|
(str(candidate_path), str(provenance_path), str(step), str(worker), maximum_candidates),
|
|
worker,
|
|
timeout_seconds,
|
|
)
|
|
if outcome == "completed":
|
|
result = read_json(worker)
|
|
worker.unlink(missing_ok=True)
|
|
return result
|
|
step.unlink(missing_ok=True)
|
|
return {
|
|
"status": "search_timeout" if outcome == "timeout" else "search_failed",
|
|
"reason": "search_worker_timeout" if outcome == "timeout" else "search_worker_failed",
|
|
}
|
|
|
|
|
|
def _public_branch_summary(result: dict[str, Any], choices: list[dict[str, Any]], comparison: dict[str, Any] | None = None) -> dict[str, Any]:
|
|
summary = {
|
|
"status": result.get("status"),
|
|
"feature_id": result.get("feature_id"),
|
|
"source_location": result.get("source_location"),
|
|
"query_group": result.get("query_group"),
|
|
"candidate_group_count": result.get("candidate_group_count"),
|
|
"reason": result.get("reason"),
|
|
"resolution_status": result.get("resolution_status"),
|
|
"choice_path": choices,
|
|
}
|
|
if comparison is not None:
|
|
summary["comparison"] = {
|
|
"decision": comparison.get("decision"),
|
|
"strict_passed": _strict_passed(comparison),
|
|
}
|
|
return {key: value for key, value in summary.items() if value is not None}
|
|
|
|
|
|
def _run_selector_search(
|
|
sample: Sample,
|
|
directory: Path,
|
|
provenance_candidate: dict[str, Any],
|
|
geometry_candidate: dict[str, Any],
|
|
gold_step: Path,
|
|
*,
|
|
rebuild_timeout_seconds: float,
|
|
comparison_timeout_seconds: float,
|
|
maximum_branches: int,
|
|
maximum_candidates: int,
|
|
) -> tuple[dict[str, Any], dict[str, Any] | None]:
|
|
"""Explore selector alternatives, accepting only a unique strict winner.
|
|
|
|
Each branch is re-executed from an untouched CDSL document. We therefore
|
|
never share mutable OCC topology between alternatives and cannot quietly
|
|
inherit the result of a previous candidate.
|
|
"""
|
|
if maximum_branches <= 0 or maximum_candidates <= 0:
|
|
return {"status": "disabled"}, None
|
|
search_dir = directory / "selector-search"
|
|
search_dir.mkdir(parents=True, exist_ok=True)
|
|
search_candidate = deepcopy(geometry_candidate)
|
|
_annotate_selector_search_tokens(search_candidate)
|
|
provenance_path = directory / "provenance.candidate.cdsl.json"
|
|
pending: list[tuple[dict[str, Any], list[dict[str, Any]]]] = [(search_candidate, [])]
|
|
seen: set[str] = set()
|
|
terminals: list[tuple[dict[str, Any], dict[str, Any], list[dict[str, Any]], dict[str, Any] | None]] = []
|
|
branch_summaries: list[dict[str, Any]] = []
|
|
budget_exhausted = False
|
|
candidates_truncated = False
|
|
branch_index = 0
|
|
while pending:
|
|
if branch_index >= maximum_branches:
|
|
budget_exhausted = True
|
|
break
|
|
candidate, choices = pending.pop(0)
|
|
fingerprint = json.dumps(candidate, sort_keys=True, separators=(",", ":"))
|
|
if fingerprint in seen:
|
|
continue
|
|
seen.add(fingerprint)
|
|
branch_index += 1
|
|
branch_name = f"branch-{branch_index:03d}"
|
|
candidate_path = search_dir / f"{branch_name}.candidate.cdsl.json"
|
|
step_path = search_dir / f"{branch_name}.rebuild.step"
|
|
write_json(candidate_path, candidate)
|
|
outcome = _run_search_replay(
|
|
candidate_path,
|
|
provenance_path,
|
|
step_path,
|
|
timeout_seconds=rebuild_timeout_seconds,
|
|
maximum_candidates=maximum_candidates,
|
|
)
|
|
_write_rebuild_artifacts(search_dir, branch_name, outcome)
|
|
if outcome.get("status") == "branchable_query_group":
|
|
groups = [item for item in outcome.get("candidate_groups") or [] if isinstance(item, dict)]
|
|
total = int(outcome.get("candidate_group_count") or len(groups))
|
|
if total > len(groups):
|
|
candidates_truncated = True
|
|
branch_summaries.append(_public_branch_summary(outcome, choices))
|
|
for group in groups:
|
|
child = deepcopy(outcome["bound_cdsl"])
|
|
feature_id = str(outcome["feature_id"])
|
|
source_feature = next(item for item in child["features"] if item.get("id") == feature_id)
|
|
token = outcome.get("selector_token")
|
|
source_selector = next(
|
|
(
|
|
selector
|
|
for _location, selector in _searchable_selector_sites(source_feature)
|
|
if selector.get(_SEARCH_TOKEN) == token
|
|
),
|
|
None,
|
|
)
|
|
if not isinstance(source_selector, dict):
|
|
raise ValueError(f"search branch selector token is unavailable: {feature_id}:{token}")
|
|
records = [item for item in group.get("records") or [] if isinstance(item, dict)]
|
|
forced = _forced_candidate_selectors(source_selector, records)
|
|
_replace_selector_group(child, feature_id, str(token), forced)
|
|
pending.append((child, [*choices, {
|
|
"feature_id": feature_id,
|
|
"source_location": outcome.get("source_location"),
|
|
"query_group": deepcopy(outcome.get("query_group") or {}),
|
|
"resolved_runtime_records": records,
|
|
"score": group.get("score"),
|
|
"ranking": deepcopy(group.get("ranking") or {}),
|
|
}]))
|
|
continue
|
|
comparison = None
|
|
if outcome.get("status") == "rebuilt":
|
|
comparison, comparison_error = _run_comparison(
|
|
gold_step,
|
|
step_path,
|
|
search_dir / f"{branch_name}.comparison.json",
|
|
timeout_seconds=comparison_timeout_seconds,
|
|
)
|
|
if comparison is not None:
|
|
write_json(search_dir / f"{branch_name}.comparison.json", comparison)
|
|
elif comparison_error is not None:
|
|
outcome["comparison_error"] = comparison_error
|
|
terminals.append((outcome, candidate, choices, comparison))
|
|
branch_summaries.append(_public_branch_summary(outcome, choices, comparison))
|
|
if pending:
|
|
budget_exhausted = True
|
|
strict_winners = [item for item in terminals if _strict_passed(item[3])]
|
|
search_complete = not budget_exhausted and not candidates_truncated
|
|
report: dict[str, Any] = {
|
|
"status": "unique_strict" if search_complete and len(strict_winners) == 1 else "no_unique_strict",
|
|
"branches_executed": branch_index,
|
|
"maximum_branches": maximum_branches,
|
|
"maximum_candidates_per_selector": maximum_candidates,
|
|
"search_complete": search_complete,
|
|
"budget_exhausted": budget_exhausted,
|
|
"candidates_truncated": candidates_truncated,
|
|
"strict_winner_count": len(strict_winners),
|
|
"branches": branch_summaries,
|
|
}
|
|
if not search_complete:
|
|
report["reason"] = "search_budget_incomplete"
|
|
return report, None
|
|
if not strict_winners:
|
|
report["reason"] = "no_strict_branch"
|
|
return report, None
|
|
if len(strict_winners) > 1:
|
|
report["reason"] = "multiple_strict_branches"
|
|
report["strict_choice_paths"] = [item[2] for item in strict_winners]
|
|
return report, None
|
|
winner, _candidate, choices, comparison = strict_winners[0]
|
|
report["strict_choice_path"] = choices
|
|
selector_record = _emit_strict_selector_record(
|
|
sample,
|
|
search_dir,
|
|
provenance_candidate,
|
|
winner,
|
|
comparison,
|
|
gold_step,
|
|
rebuild_timeout_seconds=rebuild_timeout_seconds,
|
|
comparison_timeout_seconds=comparison_timeout_seconds,
|
|
query_groups=choices,
|
|
candidate_replay_kind="selector_search_branch",
|
|
)
|
|
report["selector_record"] = selector_record
|
|
return report, selector_record if selector_record.get("status") == "strict_replayed" else None
|
|
|
|
|
|
def run_geometry_probe(
|
|
sample: Sample,
|
|
output: Path,
|
|
*,
|
|
rebuild_timeout_seconds: float = 30.0,
|
|
comparison_timeout_seconds: float = 60.0,
|
|
search_maximum_branches: int = 32,
|
|
search_maximum_candidates: int = 8,
|
|
) -> dict[str, Any]:
|
|
"""Compare normal provenance replay with an explicitly heuristic replay.
|
|
|
|
The geometry replay reuses only geometry/stable-ID fields already emitted as
|
|
diagnostic context. It never invents a selector, writes a new source
|
|
parameter, or changes the production candidate.
|
|
"""
|
|
featurescript = Path(sample.files["featurescript"])
|
|
gold_step = Path(sample.files["step"])
|
|
if not featurescript.is_file() or not gold_step.is_file():
|
|
raise FileNotFoundError(f"{sample.sample_id}: the FeatureScript and source STEP are required")
|
|
directory = output / "samples" / sample.sample_id
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
result = lower_model(parse_featurescript(featurescript.read_text(encoding="utf-8"), sample.sample_id), {})
|
|
if not isinstance(result.cdsl, dict):
|
|
raise ValueError(f"{sample.sample_id}: lowering produced no CDSL candidate")
|
|
|
|
provenance_candidate = result.cdsl
|
|
geometry_candidate = deepcopy(provenance_candidate)
|
|
removed_intents = strip_provenance_intents(geometry_candidate)
|
|
write_json(directory / "provenance.candidate.cdsl.json", provenance_candidate)
|
|
write_json(directory / "geometry_probe.candidate.cdsl.json", geometry_candidate)
|
|
|
|
provenance = _run_rebuild(
|
|
directory / "provenance.candidate.cdsl.json",
|
|
directory / "provenance.rebuild.step",
|
|
timeout_seconds=rebuild_timeout_seconds,
|
|
)
|
|
geometry_probe = _run_rebuild(
|
|
directory / "geometry_probe.candidate.cdsl.json",
|
|
directory / "geometry_probe.rebuild.step",
|
|
timeout_seconds=rebuild_timeout_seconds,
|
|
)
|
|
_write_rebuild_artifacts(directory, "provenance", provenance)
|
|
_write_rebuild_artifacts(directory, "geometry_probe", geometry_probe)
|
|
|
|
comparison = None
|
|
comparison_error = None
|
|
if geometry_probe.get("status") == "rebuilt":
|
|
comparison, comparison_error = _run_comparison(
|
|
gold_step,
|
|
directory / "geometry_probe.rebuild.step",
|
|
directory / "geometry_probe.comparison.json",
|
|
timeout_seconds=comparison_timeout_seconds,
|
|
)
|
|
if comparison is not None:
|
|
write_json(directory / "geometry_probe.comparison.json", comparison)
|
|
|
|
selector_record = _emit_strict_selector_record(
|
|
sample,
|
|
directory,
|
|
provenance_candidate,
|
|
geometry_probe,
|
|
comparison,
|
|
gold_step,
|
|
rebuild_timeout_seconds=rebuild_timeout_seconds,
|
|
comparison_timeout_seconds=comparison_timeout_seconds,
|
|
)
|
|
selector_search = {"status": "not_needed"}
|
|
if not _strict_passed(comparison):
|
|
selector_search, searched_record = _run_selector_search(
|
|
sample,
|
|
directory,
|
|
provenance_candidate,
|
|
geometry_candidate,
|
|
gold_step,
|
|
rebuild_timeout_seconds=rebuild_timeout_seconds,
|
|
comparison_timeout_seconds=comparison_timeout_seconds,
|
|
maximum_branches=search_maximum_branches,
|
|
maximum_candidates=search_maximum_candidates,
|
|
)
|
|
if searched_record is not None:
|
|
selector_record = searched_record
|
|
|
|
report = {
|
|
"schema": "cadfs_to_cdsl.selector_candidate_geometry_probe.v1",
|
|
"sample_id": sample.sample_id,
|
|
"source": {"featurescript": str(featurescript), "step": str(gold_step)},
|
|
"strategy": {
|
|
"name": "geometry_diagnostic_context",
|
|
"removed_provenance_intent_fields": removed_intents,
|
|
"classification": "heuristic_only",
|
|
"limitations": [
|
|
"The source dataset contains only a final STEP, not per-feature source checkpoints.",
|
|
"A final strict or RP pass does not prove that the FeatureScript selector is semantically correct.",
|
|
"This probe must not be used by the production resolver or counted as provenance-selector coverage.",
|
|
],
|
|
},
|
|
"provenance": _outcome_summary(provenance, None),
|
|
"geometry_probe": _outcome_summary(geometry_probe, comparison),
|
|
"selector_record": selector_record,
|
|
"selector_search": selector_search,
|
|
}
|
|
if comparison_error is not None:
|
|
report["geometry_probe"]["comparison_error"] = comparison_error
|
|
write_json(directory / "report.json", report)
|
|
return report
|
|
|
|
|
|
def _arguments() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Run an offline geometry-only selector candidate probe and strict record replay")
|
|
parser.add_argument("--input", type=Path, default=Path("data/cadfs-sample/CADFS_test"))
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument("--sample-id", action="append", required=True)
|
|
parser.add_argument("--rebuild-timeout-seconds", type=float, default=30.0)
|
|
parser.add_argument("--comparison-timeout-seconds", type=float, default=60.0)
|
|
parser.add_argument("--search-maximum-branches", type=int, default=32)
|
|
parser.add_argument("--search-maximum-candidates", type=int, default=8)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = _arguments()
|
|
samples = {sample.sample_id: sample for sample in scan_dataset(args.input, include_hashes=False)}
|
|
missing = sorted(set(args.sample_id) - samples.keys())
|
|
if missing:
|
|
raise ValueError("unknown sample ids: " + ", ".join(missing))
|
|
if args.rebuild_timeout_seconds <= 0 or args.comparison_timeout_seconds <= 0:
|
|
raise ValueError("candidate probe timeouts must be positive")
|
|
if args.search_maximum_branches < 0 or args.search_maximum_candidates < 0:
|
|
raise ValueError("search budgets cannot be negative")
|
|
reports = [
|
|
run_geometry_probe(
|
|
samples[sample_id],
|
|
args.output,
|
|
rebuild_timeout_seconds=args.rebuild_timeout_seconds,
|
|
comparison_timeout_seconds=args.comparison_timeout_seconds,
|
|
search_maximum_branches=args.search_maximum_branches,
|
|
search_maximum_candidates=args.search_maximum_candidates,
|
|
)
|
|
for sample_id in args.sample_id
|
|
]
|
|
records: list[dict[str, Any]] = []
|
|
for report in reports:
|
|
record = report.get("selector_record") or {}
|
|
record_path = record.get("record_path") if isinstance(record, dict) else None
|
|
if isinstance(record_path, str) and Path(record_path).is_file():
|
|
payload = read_json(Path(record_path))
|
|
records.extend(item for item in payload.get("records") or [] if isinstance(item, dict))
|
|
write_manifest(args.output / "selector-records.jsonl", records)
|
|
write_json(args.output / "summary.json", {
|
|
"schema": "cadfs_to_cdsl.selector_candidate_geometry_probe_summary.v2",
|
|
"sample_count": len(reports),
|
|
"strict_selector_record_count": len(records),
|
|
"reports": reports,
|
|
})
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|