1478 lines
66 KiB
Python
1478 lines
66 KiB
Python
"""Verified FeatureScript query semantics available to the selector resolver.
|
|
|
|
This is intentionally a small, explicit allow-list. A numeric source version
|
|
only identifies a FeatureScript release; it does not establish that this
|
|
runtime has verified a query family's source semantics for that release.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SelectorQueryCapability:
|
|
query_family: str
|
|
featurescript_version: str
|
|
standard_library: str
|
|
standard_library_version: str
|
|
source: str
|
|
contract: str
|
|
|
|
|
|
# These entries correspond to the direct builder contracts exercised by the
|
|
# CADFS 1511 corpus. New versions and query families must be registered with
|
|
# their source/API evidence before the runtime accepts them.
|
|
_CAPABILITIES = {
|
|
("CAP_FACE", "1511"): SelectorQueryCapability(
|
|
"CAP_FACE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history",
|
|
"direct extrude cap operation role",
|
|
),
|
|
("CAP_EDGE", "1511"): SelectorQueryCapability(
|
|
"CAP_EDGE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history",
|
|
"direct prism source edge to qualified start/end cap edge kernel lineage",
|
|
),
|
|
("CAP_EDGE", "2491"): SelectorQueryCapability(
|
|
"CAP_EDGE",
|
|
"2491",
|
|
"onshape/std/geometry.fs",
|
|
"2491.0",
|
|
"CADFS FeatureScript 2491 symmetric extrusion query history",
|
|
"immediate direct two-sided prism source edge to qualified start/end cap edge kernel lineage",
|
|
),
|
|
("CAP_VERTEX", "1511"): SelectorQueryCapability(
|
|
"CAP_VERTEX",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history",
|
|
"direct new-body prism source-vertex to qualified start/end cap vertex kernel lineage",
|
|
),
|
|
("OFFSET_FACE", "1511"): SelectorQueryCapability(
|
|
"OFFSET_FACE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history",
|
|
"shell offset-face operation role with true dependency qualification",
|
|
),
|
|
("OFFSET_EDGE", "1511"): SelectorQueryCapability(
|
|
"OFFSET_EDGE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history",
|
|
"one-sided direct-prism retained cap edge through an immediate shell continuation",
|
|
),
|
|
# These entries authorize the deliberately narrow direct-prism lineage
|
|
# path. The capability matrix keeps the broader query families explicitly
|
|
# partial; all other generator and lifecycle combinations remain rejected.
|
|
("SWEPT_FACE", "1511"): SelectorQueryCapability(
|
|
"SWEPT_FACE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history",
|
|
"kernel-lineage resolver contract only",
|
|
),
|
|
("SWEPT_EDGE", "1511"): SelectorQueryCapability(
|
|
"SWEPT_EDGE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history",
|
|
"kernel-lineage resolver contract only",
|
|
),
|
|
("SWEPT_EDGE", "2491"): SelectorQueryCapability(
|
|
"SWEPT_EDGE",
|
|
"2491",
|
|
"onshape/std/geometry.fs",
|
|
"2491.0",
|
|
"CADFS FeatureScript 2491 exported full-revolve query history",
|
|
"full independent solid revolve source vertex to exact MakeRevol Generated(vertex) kernel lineage",
|
|
),
|
|
("SWEPT_BODY", "1511"): SelectorQueryCapability(
|
|
"SWEPT_BODY",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history; one-item qUnion identity",
|
|
"active direct-new-body member selected by its producing operation",
|
|
),
|
|
("INTERSECT", "1511"): SelectorQueryCapability(
|
|
"INTERSECT",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported query history; BRepAlgoAPI boolean Generated(face) and SectionEdges() exact handles",
|
|
"two source-qualified boolean input faces to one final section edge kernel lineage",
|
|
),
|
|
("BLEND_EDGE", "1511"): SelectorQueryCapability(
|
|
"BLEND_EDGE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported direct dress-up query history",
|
|
"one direct-prism CAP_EDGE and CAP_FACE source pair to one exact final fillet/chamfer patch boundary",
|
|
),
|
|
("BLEND_FACE", "1511"): SelectorQueryCapability(
|
|
"BLEND_FACE",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 exported direct dress-up sketch history",
|
|
"one immediate direct-prism CAP_EDGE to its exact active native fillet/chamfer generated patch face",
|
|
),
|
|
("COPY", "1511"): SelectorQueryCapability(
|
|
"COPY",
|
|
"1511",
|
|
"onshape/std/geometry.fs",
|
|
"1511.0",
|
|
"CADFS FeatureScript 1511 primary boolean COPY query history",
|
|
"immediate primary-cut COPY(CAP_EDGE) projection over an exact transient-prism and cut lineage",
|
|
),
|
|
}
|
|
|
|
|
|
def selector_query_capability(intent: dict[str, Any]) -> SelectorQueryCapability | None:
|
|
"""Return an explicitly verified query capability for ``intent`` only."""
|
|
source_query = intent.get("source_query")
|
|
if not isinstance(source_query, dict):
|
|
return None
|
|
family = intent.get("query_family")
|
|
version = source_query.get("featurescript_version")
|
|
if not isinstance(family, str) or not isinstance(version, str):
|
|
return None
|
|
capability = _CAPABILITIES.get((family, version))
|
|
if capability is None:
|
|
return None
|
|
if (
|
|
source_query.get("standard_library") != capability.standard_library
|
|
or source_query.get("standard_library_version") != capability.standard_library_version
|
|
):
|
|
return None
|
|
return capability
|
|
|
|
|
|
def known_selector_query_versions(query_family: str) -> tuple[str, ...]:
|
|
"""Expose registered versions for deterministic unsupported diagnostics."""
|
|
return tuple(sorted(version for family, version in _CAPABILITIES if family == query_family))
|
|
|
|
|
|
_PROVEN_OPERAND_SET_CONTRACTS = {
|
|
"proven_operand_union": ("union", "qUnion", 2, None),
|
|
"proven_operand_intersection": ("intersection", "qIntersection", 2, None),
|
|
"proven_operand_subtraction": ("subtraction", "qSubtraction", 2, 2),
|
|
}
|
|
|
|
|
|
def proven_operand_set_contract_error(selector: dict[str, Any]) -> str | None:
|
|
"""Validate a narrowly executable FeatureScript query-set contract.
|
|
|
|
The parent is a set expression, not another geometric/topology lookup.
|
|
Each ordered child therefore has to be a fully expressed provenance
|
|
selector whose typed expression is exactly the matching source-AST leaf.
|
|
The direct bridge deliberately requires each child to resolve to a
|
|
non-empty proven active set. It does not mistake an absent lineage for a
|
|
mathematically empty FeatureScript operand.
|
|
"""
|
|
intent = selector.get("selector_intent")
|
|
has_set_fields = (
|
|
selector.get("query_operands") is not None
|
|
or isinstance(intent, dict) and intent.get("query_set_contract") is not None
|
|
or isinstance(intent, dict) and intent.get("query_family") == "QUERY_SET"
|
|
)
|
|
if not has_set_fields:
|
|
return None
|
|
if not isinstance(intent, dict):
|
|
return "query-set selector requires selector_intent"
|
|
if intent.get("query_family") != "QUERY_SET":
|
|
return "query-set selector intent must use query_family QUERY_SET"
|
|
contract = intent.get("query_set_contract")
|
|
contract_spec = _PROVEN_OPERAND_SET_CONTRACTS.get(contract)
|
|
if contract_spec is None:
|
|
return "query-set selector has an unsupported set contract"
|
|
operator, source_name, minimum_operands, maximum_operands = contract_spec
|
|
kind = selector.get("kind")
|
|
if kind not in {"face", "edge"} or intent.get("set_kind") != kind:
|
|
return "query-set selector kind must be an explicit face or edge set kind"
|
|
if (
|
|
selector.get("source") != "runtime_snapshot"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or intent.get("body_scope") != "active_member"
|
|
or intent.get("empty_policy") != "reject"
|
|
or intent.get("multiple_policy") != "all"
|
|
):
|
|
return "query-set selector does not declare the active-member set policy"
|
|
if any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id",
|
|
"owner_feature_id", "output_role", "output_role_source",
|
|
"matched_selectors", "intersection_of",
|
|
)):
|
|
return "query-set selector cannot mix stable, output-role, or geometric evidence"
|
|
policy = intent.get("derivation_policy")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "source_qualified"
|
|
or not isinstance(policy.get("allowed"), list)
|
|
or not policy["allowed"]
|
|
):
|
|
return "query-set selector has an invalid set derivation policy"
|
|
expression = intent.get("query_expr")
|
|
root = expression.get("root") if isinstance(expression, dict) else None
|
|
operands = selector.get("query_operands")
|
|
if (
|
|
not isinstance(root, dict)
|
|
or root.get("node") != "set"
|
|
or root.get("operator") != operator
|
|
or not isinstance(root.get("operands"), list)
|
|
or not isinstance(operands, list)
|
|
or len(root["operands"]) < minimum_operands
|
|
or len(root["operands"]) != len(operands)
|
|
or maximum_operands is not None and len(root["operands"]) != maximum_operands
|
|
):
|
|
return f"{source_name} selector operands do not match its source expression"
|
|
source_query = intent.get("source_query")
|
|
source_signature = (
|
|
source_query.get("featurescript_version"),
|
|
source_query.get("standard_library"),
|
|
source_query.get("standard_library_version"),
|
|
) if isinstance(source_query, dict) else None
|
|
def validate_child(
|
|
operand: Any,
|
|
expression_leaf: Any,
|
|
*,
|
|
index_path: str,
|
|
) -> str | None:
|
|
"""Validate one recursive child against its exact source AST node."""
|
|
if not isinstance(operand, dict) or operand.get("kind") != kind:
|
|
return f"{source_name} operand {index_path} does not have the parent set kind"
|
|
if (
|
|
operand.get("source") != "runtime_snapshot"
|
|
or any(operand.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id",
|
|
"matched_selectors", "intersection_of",
|
|
))
|
|
):
|
|
return f"{source_name} operand {index_path} has unsupported non-provenance evidence"
|
|
operand_intent = operand.get("selector_intent")
|
|
if not isinstance(operand_intent, dict):
|
|
return f"{source_name} operand {index_path} is missing selector intent"
|
|
operand_expression = operand_intent.get("query_expr")
|
|
if not isinstance(operand_expression, dict) or operand_expression.get("root") != expression_leaf:
|
|
return f"{source_name} operand {index_path} does not match its source expression leaf"
|
|
operand_source = operand_intent.get("source_query")
|
|
operand_signature = (
|
|
operand_source.get("featurescript_version"),
|
|
operand_source.get("standard_library"),
|
|
operand_source.get("standard_library_version"),
|
|
) if isinstance(operand_source, dict) else None
|
|
if operand_signature != source_signature:
|
|
return f"{source_name} operand {index_path} has incompatible FeatureScript source metadata"
|
|
|
|
nested_operands = operand.get("query_operands")
|
|
if nested_operands is not None or operand_intent.get("query_family") == "QUERY_SET":
|
|
nested_error = proven_operand_set_contract_error(operand)
|
|
if nested_error is not None:
|
|
return f"{source_name} operand {index_path} has invalid nested query set: {nested_error}"
|
|
return None
|
|
|
|
if operand_intent.get("query_family") in {None, "GEOMETRIC"}:
|
|
return f"{source_name} operand {index_path} is not a direct provenance selector"
|
|
operand_policy = operand_intent.get("derivation_policy")
|
|
if not isinstance(operand_policy, dict) or operand_policy.get("multiplicity") == "none":
|
|
return f"{source_name} operand {index_path} is not executable provenance"
|
|
return None
|
|
|
|
for index, (operand, expression_leaf) in enumerate(zip(operands, root["operands"])):
|
|
error = validate_child(operand, expression_leaf, index_path=str(index))
|
|
if error is not None:
|
|
return error
|
|
return None
|
|
|
|
|
|
def blend_face_selector_contract_error(selector: dict[str, Any]) -> str | None:
|
|
"""Validate the narrow runtime-attached direct-prism BLEND_FACE form."""
|
|
intent = selector.get("selector_intent")
|
|
if not isinstance(intent, dict) or intent.get("query_family") != "BLEND_FACE":
|
|
return None
|
|
# Preserve generic source queries as deferred diagnostics. Only the
|
|
# explicitly declared direct-prism contract is executable.
|
|
if intent.get("blend_face_source") is None:
|
|
return None
|
|
source = intent.get("blend_face_source")
|
|
policy = intent.get("derivation_policy")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or intent.get("kind") != "face"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or not isinstance(policy, dict)
|
|
or policy.get("allowed") != ["boundary"]
|
|
or policy.get("multiplicity") != "one"
|
|
or not isinstance(source, dict)
|
|
or source.get("query_family") != "CAP_EDGE"
|
|
or not isinstance(source.get("owner_feature_id"), str)
|
|
or not isinstance(source.get("source_entity"), dict)
|
|
or source.get("lineage_role") not in {"extrude.start", "extrude.end"}
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id",
|
|
"output_role", "query_input", "query_operands", "matched_selectors", "intersection_of",
|
|
))
|
|
):
|
|
return "BLEND_FACE selector does not declare a supported direct-prism patch contract"
|
|
source_entity = source["source_entity"]
|
|
if not all(isinstance(source_entity.get(key), str) and source_entity[key] for key in ("sketch_id", "entity_id")):
|
|
return "BLEND_FACE CAP_EDGE source is incomplete"
|
|
return None
|
|
|
|
|
|
def copy_selector_contract_error(selector: dict[str, Any]) -> str | None:
|
|
"""Validate the narrow source-qualified primary-cut COPY projection."""
|
|
intent = selector.get("selector_intent")
|
|
if not isinstance(intent, dict) or intent.get("query_family") != "COPY":
|
|
return None
|
|
# A general COPY query remains deliberately deferred. Validate only the
|
|
# explicit contract form so it retains its existing unsupported diagnostic
|
|
# instead of being misreported as a malformed executable bridge.
|
|
if intent.get("copy_contract") is None and selector.get("query_input") is None:
|
|
return None
|
|
copy_contract = intent.get("copy_contract")
|
|
face_contracts = {
|
|
"primary_cut_cap_face_workplane": "CAP_FACE",
|
|
"primary_cut_swept_face_workplane": "SWEPT_FACE",
|
|
}
|
|
expected_kind = "face" if copy_contract in face_contracts else "edge"
|
|
expected_input_family = face_contracts.get(copy_contract, "CAP_EDGE")
|
|
if (
|
|
selector.get("kind") != expected_kind
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or intent.get("kind") != expected_kind
|
|
or intent.get("evidence") != "kernel_history"
|
|
or copy_contract not in {"primary_cut_cap_edge", *face_contracts}
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id",
|
|
"output_role", "query_operands", "matched_selectors", "intersection_of",
|
|
))
|
|
):
|
|
return "COPY selector does not declare a supported primary-cut contract"
|
|
policy = intent.get("derivation_policy")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("allowed") != ["boundary", "continuation"]
|
|
or policy.get("multiplicity") != "one"
|
|
):
|
|
return "COPY selector has an invalid primary-cut derivation policy"
|
|
query_input = selector.get("query_input")
|
|
input_intent = query_input.get("selector_intent") if isinstance(query_input, dict) else None
|
|
if (
|
|
not isinstance(query_input, dict)
|
|
or query_input.get("kind") != expected_kind
|
|
or query_input.get("owner_feature_id") != selector.get("owner_feature_id")
|
|
or query_input.get("source") != "runtime_snapshot"
|
|
or not isinstance(input_intent, dict)
|
|
or input_intent.get("query_family") != expected_input_family
|
|
or input_intent.get("evidence") != "kernel_history"
|
|
or (
|
|
copy_contract != "primary_cut_swept_face_workplane"
|
|
and input_intent.get("lineage_role") not in {"extrude.start", "extrude.end"}
|
|
)
|
|
):
|
|
return "COPY selector requires one same-owner primary-cut query input"
|
|
if expected_kind == "edge" and not isinstance(input_intent.get("source_entity"), dict):
|
|
return "COPY CAP_EDGE selector requires one source-profile edge"
|
|
if copy_contract == "primary_cut_cap_face_workplane" and (not isinstance(input_intent.get("source_entities"), list) or not input_intent["source_entities"]):
|
|
return "COPY CAP_FACE selector requires its complete source-profile edge set"
|
|
if copy_contract == "primary_cut_swept_face_workplane" and not isinstance(input_intent.get("source_entity"), dict):
|
|
return "COPY SWEPT_FACE selector requires one source-profile edge"
|
|
input_policy = input_intent.get("derivation_policy")
|
|
if (
|
|
not isinstance(input_policy, dict)
|
|
or input_policy.get("allowed") != ["boundary", "continuation"]
|
|
or input_policy.get("multiplicity") != "one"
|
|
):
|
|
return "COPY selector input has an invalid primary-cut derivation policy"
|
|
source_query = intent.get("source_query")
|
|
input_source = input_intent.get("source_query")
|
|
signature = (
|
|
source_query.get("featurescript_version"),
|
|
source_query.get("standard_library"),
|
|
source_query.get("standard_library_version"),
|
|
) if isinstance(source_query, dict) else None
|
|
input_signature = (
|
|
input_source.get("featurescript_version"),
|
|
input_source.get("standard_library"),
|
|
input_source.get("standard_library_version"),
|
|
) if isinstance(input_source, dict) else None
|
|
if signature != input_signature:
|
|
return "COPY selector input has incompatible FeatureScript source metadata"
|
|
expression = intent.get("query_expr")
|
|
input_expression = input_intent.get("query_expr")
|
|
root = expression.get("root") if isinstance(expression, dict) else None
|
|
input_root = input_expression.get("root") if isinstance(input_expression, dict) else None
|
|
# CADFS sketches commonly wrap one workplane query in qUnion([Q0]). That
|
|
# is a set-preserving identity wrapper, not a different COPY relation.
|
|
copy_root = root
|
|
if (
|
|
isinstance(copy_root, dict)
|
|
and copy_root.get("node") == "set"
|
|
and copy_root.get("operator") == "union"
|
|
and isinstance(copy_root.get("operands"), list)
|
|
and len(copy_root["operands"]) == 1
|
|
):
|
|
copy_root = copy_root["operands"][0]
|
|
if (
|
|
not isinstance(copy_root, dict)
|
|
or copy_root.get("node") != "topology_query"
|
|
or copy_root.get("topology_type") != {"node": "literal", "value": "COPY"}
|
|
or not isinstance(copy_root.get("arguments"), list)
|
|
or len(copy_root["arguments"]) != 1
|
|
or not isinstance(copy_root["arguments"][0], dict)
|
|
or copy_root["arguments"][0].get("node") != "map"
|
|
or not isinstance(input_root, dict)
|
|
):
|
|
return "COPY selector has no typed COPY source expression"
|
|
derived = [
|
|
entry.get("value")
|
|
for entry in copy_root["arguments"][0].get("entries") or ()
|
|
if isinstance(entry, dict) and entry.get("key") == "derivedFrom"
|
|
]
|
|
if len(derived) != 1 or derived[0] != input_root:
|
|
return "COPY selector input does not match the COPY derivedFrom expression"
|
|
return None
|
|
|
|
|
|
def owner_body_selector_contract_error(selector: dict[str, Any]) -> str | None:
|
|
"""Validate the narrow exact-input ``qOwnerBody`` contract.
|
|
|
|
``qOwnerBody`` is an ownership projection, not a request to search the
|
|
current aggregate for a body. The only executable form therefore keeps
|
|
one nested, already-proven topology selector and asks the runtime to
|
|
project its exact active ``body_id`` to the matching body record.
|
|
"""
|
|
intent = selector.get("selector_intent")
|
|
if not isinstance(intent, dict) or intent.get("query_family") != "OWNER_BODY":
|
|
return None
|
|
if (
|
|
selector.get("kind") != "body"
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or intent.get("kind") != "body"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or intent.get("owner_body_contract") != "exact_input_owner"
|
|
or intent.get("body_scope") != "active_member"
|
|
or intent.get("empty_policy") != "reject"
|
|
or intent.get("multiple_policy") != "one"
|
|
or selector.get("output_role") is not None
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id",
|
|
"matched_selectors", "intersection_of", "query_operands",
|
|
))
|
|
):
|
|
return "qOwnerBody selector does not declare the exact active-member owner contract"
|
|
policy = intent.get("derivation_policy")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or policy.get("allowed") != ["boundary"]
|
|
):
|
|
return "qOwnerBody selector has an invalid owner derivation policy"
|
|
expression = intent.get("query_expr")
|
|
root = expression.get("root") if isinstance(expression, dict) else None
|
|
if (
|
|
not isinstance(root, dict)
|
|
or root.get("node") != "filter"
|
|
or root.get("filter") != "owner_body"
|
|
or not isinstance(root.get("input"), dict)
|
|
or not isinstance(selector.get("query_input"), dict)
|
|
):
|
|
return "qOwnerBody selector does not retain one typed query input"
|
|
query_input = selector["query_input"]
|
|
if query_input.get("kind") == "body":
|
|
return "qOwnerBody input must name a topology member, not a body"
|
|
input_intent = query_input.get("selector_intent")
|
|
if not isinstance(input_intent, dict):
|
|
return "qOwnerBody input is missing selector intent"
|
|
input_expression = input_intent.get("query_expr")
|
|
if not isinstance(input_expression, dict) or input_expression.get("root") != root.get("input"):
|
|
return "qOwnerBody input does not match its source expression"
|
|
source_query = intent.get("source_query")
|
|
input_source = input_intent.get("source_query")
|
|
source_signature = (
|
|
source_query.get("featurescript_version"),
|
|
source_query.get("standard_library"),
|
|
source_query.get("standard_library_version"),
|
|
) if isinstance(source_query, dict) else None
|
|
input_signature = (
|
|
input_source.get("featurescript_version"),
|
|
input_source.get("standard_library"),
|
|
input_source.get("standard_library_version"),
|
|
) if isinstance(input_source, dict) else None
|
|
if source_signature != input_signature:
|
|
return "qOwnerBody input has incompatible FeatureScript source metadata"
|
|
if (
|
|
query_input.get("query_operands") is not None
|
|
or input_intent.get("query_family") in {None, "GEOMETRIC", "OWNER_BODY", "QUERY_SET"}
|
|
):
|
|
return "qOwnerBody input is not a proven topology selector"
|
|
input_policy = input_intent.get("derivation_policy")
|
|
if not isinstance(input_policy, dict) or input_policy.get("multiplicity") == "none":
|
|
return "qOwnerBody input is not executable provenance"
|
|
return None
|
|
|
|
|
|
def known_selector_query_standard_library_versions(query_family: str) -> tuple[tuple[str, str], ...]:
|
|
"""Expose exact direct imports that back a registered query contract."""
|
|
return tuple(sorted({
|
|
(capability.standard_library, capability.standard_library_version)
|
|
for (family, _version), capability in _CAPABILITIES.items()
|
|
if family == query_family
|
|
}))
|
|
|
|
|
|
def _direct_profile_source_entity_ids(sketch: dict[str, Any]) -> set[str]:
|
|
"""Return source labels retained by one unchanged direct profile."""
|
|
profile = sketch.get("profile") or {}
|
|
source_ids: set[str] = set()
|
|
direct_circle = profile.get("source_entity_id") if profile.get("type") == "circle" else None
|
|
if isinstance(direct_circle, str) and direct_circle:
|
|
source_ids.add(direct_circle)
|
|
for contour in profile.get("contours") or ():
|
|
if not isinstance(contour, dict):
|
|
continue
|
|
for segment in contour.get("segments") or ():
|
|
source_entity_id = segment.get("source_entity_id") if isinstance(segment, dict) else None
|
|
if isinstance(source_entity_id, str) and source_entity_id:
|
|
source_ids.add(source_entity_id)
|
|
return source_ids
|
|
|
|
|
|
def _has_one_exact_retained_source_edge(
|
|
selected: dict[str, Any],
|
|
source: dict[str, Any],
|
|
source_entity_id: str,
|
|
) -> bool:
|
|
"""Prove one selected prism edge is unchanged from its source sketch."""
|
|
selected_profile = selected.get("profile") or {}
|
|
source_profile = source.get("profile") or {}
|
|
if selected.get("workplane") != source.get("workplane"):
|
|
return False
|
|
|
|
def segments(profile: dict[str, Any]) -> list[dict[str, Any]]:
|
|
if profile.get("type") != "analytic_contours":
|
|
return []
|
|
return [
|
|
segment
|
|
for contour in profile.get("contours") or ()
|
|
if isinstance(contour, dict)
|
|
for segment in contour.get("segments") or ()
|
|
if isinstance(segment, dict) and segment.get("source_entity_id") == source_entity_id
|
|
]
|
|
|
|
selected_segments = segments(selected_profile)
|
|
source_segments = segments(source_profile)
|
|
return len(selected_segments) == len(source_segments) == 1 and selected_segments[0] == source_segments[0]
|
|
|
|
|
|
def is_immediate_retained_source_prism_swept_face_extent(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the immediate IMPRINT-prism wall extent contract."""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
source_entity = intent.get("source_entity") if isinstance(intent, dict) else None
|
|
policy = intent.get("derivation_policy") if isinstance(intent, dict) else None
|
|
source_query = intent.get("source_query") if isinstance(intent, dict) else None
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("owner_feature_id") != producer.get("id")
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role") is not None
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id", "output_role_source",
|
|
))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "SWEPT_FACE"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or intent.get("consumer_contract") != "immediate_retained_source_prism_swept_face_up_to_surface"
|
|
or not isinstance(policy, dict)
|
|
or policy != {"allowed": ["boundary"], "multiplicity": "one"}
|
|
or not isinstance(source_query, dict)
|
|
or source_query.get("featurescript_version") != "1511"
|
|
or source_query.get("standard_library") != "onshape/std/geometry.fs"
|
|
or source_query.get("standard_library_version") != "1511.0"
|
|
or not isinstance(source_entity, dict)
|
|
):
|
|
return False
|
|
sketch_id = source_entity.get("sketch_id")
|
|
entity_id = source_entity.get("entity_id")
|
|
params = producer.get("params") or {}
|
|
selected = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
source = next(
|
|
(sketch for sketch in sketches.values() if sketch.get("source_sketch_id") == sketch_id),
|
|
{},
|
|
)
|
|
return (
|
|
producer.get("atomic_id") == "extrude_add_blind"
|
|
and params.get("result_mode") == "new_body"
|
|
and (params.get("end_condition") or {}).get("type") == "blind"
|
|
and params.get("draft") is None
|
|
and isinstance(sketch_id, str)
|
|
and isinstance(entity_id, str)
|
|
and selected.get("source_sketch_id") == sketch_id
|
|
and _has_one_exact_retained_source_edge(selected, source, entity_id)
|
|
)
|
|
|
|
|
|
def is_direct_blind_extrude_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Whether a selector names one cap from the direct prism contract.
|
|
|
|
This is the consumer-side mirror of the lowerer's CAP_FACE constructor.
|
|
It deliberately describes a narrow builder contract, rather than treating
|
|
an output-role string as general permission to select arbitrary topology.
|
|
The runtime still has to prove the exact role in its active snapshot.
|
|
"""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("output_role") not in {"extrude.start", "extrude.end"}
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role_source") is not None
|
|
or any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id"))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_FACE"
|
|
or intent.get("evidence") != "operation_role"
|
|
or intent.get("output_role") != selector.get("output_role")
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
if not isinstance(policy, dict) or policy.get("multiplicity") != "one" or set(policy.get("allowed") or ()) != {"boundary", "continuation"}:
|
|
return False
|
|
params = producer.get("params") or {}
|
|
if (
|
|
producer.get("atomic_id") != "extrude_add_blind"
|
|
or params.get("result_mode") != "new_body"
|
|
or (params.get("end_condition") or {}).get("type") != "blind"
|
|
):
|
|
return False
|
|
if params.get("draft") is None:
|
|
return True
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
profile = sketch.get("profile") or {}
|
|
if profile.get("type") == "circle":
|
|
return True
|
|
contours = profile.get("contours")
|
|
return (
|
|
profile.get("type") == "analytic_contours"
|
|
and isinstance(contours, list)
|
|
and len(contours) == 1
|
|
and bool((contours[0] or {}).get("closed"))
|
|
)
|
|
|
|
|
|
def is_direct_prism_shell_offset_edge_tdd(
|
|
selector: dict[str, Any],
|
|
shell: dict[str, Any] | None,
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the retained-cap ``OFFSET_EDGE`` TDD tuple.
|
|
|
|
This is intentionally not a shell-wall role bridge. In this source form
|
|
the outer shell query's TDD names one retained direct-prism CAP_EDGE. OCC
|
|
proves it through the cap-edge boundary followed by the shell's exact
|
|
one-to-one continuation. The deleted opposite cap can generate an inner
|
|
shell wall, but that is a different result and cannot satisfy this query.
|
|
"""
|
|
if not isinstance(selector, dict) or not isinstance(shell, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
if (
|
|
selector.get("kind") != "edge"
|
|
or selector.get("owner_feature_id") != producer.get("id")
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id", "output_role", "output_role_source",
|
|
))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "OFFSET_EDGE"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or intent.get("consumer_contract") != "direct_prism_shell_offset_edge_tdd"
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
source_entity = intent.get("source_entity")
|
|
disambiguation = intent.get("disambiguation")
|
|
role = intent.get("lineage_role")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or set(policy.get("allowed") or ()) != {"boundary", "continuation"}
|
|
or not isinstance(source_entity, dict)
|
|
or not isinstance(source_entity.get("sketch_id"), str)
|
|
or not isinstance(source_entity.get("entity_id"), str)
|
|
or role not in {"extrude.start", "extrude.end"}
|
|
or not isinstance(disambiguation, dict)
|
|
or disambiguation.get("type") != "offset_edge_tdd_cap_continuation"
|
|
or disambiguation.get("shell_feature_id") != shell.get("id")
|
|
or disambiguation.get("outer_owner_feature_id") != shell.get("id")
|
|
or disambiguation.get("tdd_cap_owner_feature_id") != producer.get("id")
|
|
or disambiguation.get("tdd_cap_role") != role
|
|
or disambiguation.get("source_entity") != source_entity
|
|
):
|
|
return False
|
|
params = producer.get("params") or {}
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
if (
|
|
producer.get("atomic_id") != "extrude_add_blind"
|
|
or params.get("result_mode") != "new_body"
|
|
or (params.get("end_condition") or {}).get("type") != "blind"
|
|
or params.get("draft") is not None
|
|
or sketch.get("source_sketch_id") != source_entity["sketch_id"]
|
|
or source_entity["entity_id"] not in _direct_profile_source_entity_ids(sketch)
|
|
):
|
|
return False
|
|
shell_params = shell.get("params") or {}
|
|
shell_selectors = shell.get("selectors") or []
|
|
if (
|
|
shell.get("atomic_id") != "shell"
|
|
or shell.get("depends_on") != [producer.get("id")]
|
|
or shell_params.get("inward") is not True
|
|
or len(shell_selectors) != 1
|
|
or not isinstance(shell_selectors[0], dict)
|
|
):
|
|
return False
|
|
removed = shell_selectors[0]
|
|
expected_removed = "extrude.end" if role == "extrude.start" else "extrude.start"
|
|
removed_intent = removed.get("selector_intent")
|
|
return (
|
|
removed.get("kind") == "face"
|
|
and removed.get("owner_feature_id") == producer.get("id")
|
|
and removed.get("output_role") == expected_removed
|
|
and removed.get("source") == "runtime_snapshot"
|
|
and isinstance(removed_intent, dict)
|
|
and removed_intent.get("query_family") == "CAP_FACE"
|
|
and removed_intent.get("output_role") == expected_removed
|
|
)
|
|
|
|
|
|
def is_direct_prism_shell_offset_edge_vertex(
|
|
selector: dict[str, Any],
|
|
shell: dict[str, Any] | None,
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the OSD-only source-vertex shell ``OFFSET_EDGE`` tuple."""
|
|
if not isinstance(selector, dict) or not isinstance(shell, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
if (
|
|
selector.get("kind") != "edge"
|
|
or selector.get("owner_feature_id") != producer.get("id")
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id", "output_role", "output_role_source",
|
|
))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "OFFSET_EDGE"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or intent.get("consumer_contract") != "direct_prism_shell_offset_edge_vertex"
|
|
or intent.get("source_entity") is not None
|
|
or intent.get("lineage_role") is not None
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
sources = intent.get("source_entities")
|
|
disambiguation = intent.get("disambiguation")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or set(policy.get("allowed") or ()) != {"boundary", "continuation"}
|
|
or not isinstance(sources, list) or len(sources) != 2
|
|
or not all(
|
|
isinstance(source, dict)
|
|
and isinstance(source.get("sketch_id"), str) and source["sketch_id"]
|
|
and isinstance(source.get("entity_id"), str) and source["entity_id"]
|
|
for source in sources
|
|
)
|
|
or len({(source["sketch_id"], source["entity_id"]) for source in sources}) != 2
|
|
or not isinstance(disambiguation, dict)
|
|
or disambiguation.get("type") != "offset_edge_vertex_continuation"
|
|
or disambiguation.get("shell_feature_id") != shell.get("id")
|
|
or disambiguation.get("outer_owner_feature_id") != shell.get("id")
|
|
or disambiguation.get("prism_owner_feature_id") != producer.get("id")
|
|
or disambiguation.get("source_entities") != sources
|
|
):
|
|
return False
|
|
source_sketches = {source["sketch_id"] for source in sources}
|
|
entity_ids = {source["entity_id"] for source in sources}
|
|
params = producer.get("params") or {}
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
if (
|
|
len(source_sketches) != 1
|
|
or sketch.get("source_sketch_id") != next(iter(source_sketches))
|
|
or not entity_ids <= _direct_profile_source_entity_ids(sketch)
|
|
or producer.get("atomic_id") != "extrude_add_blind"
|
|
or params.get("result_mode") != "new_body"
|
|
or (params.get("end_condition") or {}).get("type") != "blind"
|
|
or params.get("draft") is not None
|
|
):
|
|
return False
|
|
shell_params = shell.get("params") or {}
|
|
shell_selectors = shell.get("selectors") or []
|
|
if (
|
|
shell.get("atomic_id") != "shell"
|
|
or shell.get("depends_on") != [producer.get("id")]
|
|
or shell_params.get("inward") is not True
|
|
or len(shell_selectors) != 1
|
|
or not isinstance(shell_selectors[0], dict)
|
|
):
|
|
return False
|
|
removed = shell_selectors[0]
|
|
removed_intent = removed.get("selector_intent")
|
|
return (
|
|
removed.get("kind") == "face"
|
|
and removed.get("owner_feature_id") == producer.get("id")
|
|
and removed.get("output_role") in {"extrude.start", "extrude.end"}
|
|
and removed.get("output_role") == disambiguation.get("removed_cap_role")
|
|
and removed.get("source") == "runtime_snapshot"
|
|
and isinstance(removed_intent, dict)
|
|
and removed_intent.get("query_family") == "CAP_FACE"
|
|
and removed_intent.get("output_role") == removed.get("output_role")
|
|
)
|
|
|
|
|
|
def is_primary_add_shell_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
_sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the primary-ADD CAP role available only to an immediate shell.
|
|
|
|
Unlike a ``new_body`` prism, this producer's tool is transient. The
|
|
executor therefore has to prove its cap's one-to-one union successor in
|
|
the final active member; this predicate only admits the source/CDSL
|
|
contract needed to request that proof.
|
|
"""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("output_role") not in {"extrude.start", "extrude.end"}
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role_source") is not None
|
|
or any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id"))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_FACE"
|
|
or intent.get("evidence") != "operation_role"
|
|
or intent.get("output_role") != selector.get("output_role")
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or set(policy.get("allowed") or ()) != {"boundary", "continuation"}
|
|
):
|
|
return False
|
|
params = producer.get("params") or {}
|
|
return (
|
|
producer.get("atomic_id") == "extrude_add_blind"
|
|
and params.get("result_mode") != "new_body"
|
|
and (params.get("end_condition") or {}).get("type") == "blind"
|
|
and params.get("draft") is None
|
|
)
|
|
|
|
|
|
def is_primary_add_up_to_surface_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the transient primary-ADD CAP role for an immediate extent.
|
|
|
|
This has the same producer constraints as the shell bridge, but is kept
|
|
separately named because its consumer must calculate a true face target,
|
|
not remove the face itself. The active successor remains subject to the
|
|
exact union continuation proof in ``TopologyRegistry.resolve``.
|
|
"""
|
|
intent = selector.get("selector_intent") if isinstance(selector, dict) else None
|
|
return (
|
|
isinstance(intent, dict)
|
|
and intent.get("consumer_contract") == "primary_add_up_to_surface_union_continuation"
|
|
and is_primary_add_shell_cap_output_role(selector, producer, sketches)
|
|
)
|
|
|
|
|
|
def is_primary_add_dressup_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the transient primary-ADD CAP role for an immediate dress-up.
|
|
|
|
The consuming fillet/chamfer expands the resolved physical face to its
|
|
actual body-boundary edges. It does not infer those edges from source
|
|
geometry: the output role must first survive the exact union continuation.
|
|
"""
|
|
intent = selector.get("selector_intent") if isinstance(selector, dict) else None
|
|
return (
|
|
isinstance(intent, dict)
|
|
and intent.get("consumer_contract") == "primary_add_dressup_union_continuation"
|
|
and is_primary_add_shell_cap_output_role(selector, producer, sketches)
|
|
)
|
|
|
|
|
|
def is_symmetric_direct_prism_two_sided_up_to_surface_cap_pair(
|
|
forward: dict[str, Any],
|
|
reverse: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the only paired CAP contract for two-sided up-to-surface.
|
|
|
|
Both directions are required because a symmetric prism has two far caps
|
|
and no source-plane cap. Each query must name the complete, unchanged
|
|
direct source profile; this keeps a partial OSD query from becoming a
|
|
geometric substitute for a CAP role.
|
|
"""
|
|
if not isinstance(forward, dict) or not isinstance(reverse, dict) or not isinstance(producer, dict):
|
|
return False
|
|
selectors = (forward, reverse)
|
|
contract = "symmetric_direct_prism_two_sided_up_to_surface_cap_pair"
|
|
for selector in selectors:
|
|
intent = selector.get("selector_intent")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("output_role") not in {"extrude.start", "extrude.end"}
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role_source") is not None
|
|
or any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id"))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_FACE"
|
|
or intent.get("evidence") != "operation_role"
|
|
or intent.get("output_role") != selector.get("output_role")
|
|
or intent.get("consumer_contract") != contract
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
source = intent.get("source_query")
|
|
ids = (intent.get("disambiguation") or {}).get("source_profile_entity_ids")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or set(policy.get("allowed") or ()) != {"boundary"}
|
|
or not isinstance(source, dict)
|
|
or source.get("featurescript_version") != "1511"
|
|
or source.get("standard_library") != "onshape/std/geometry.fs"
|
|
or source.get("standard_library_version") != "1511.0"
|
|
or not isinstance(ids, list)
|
|
or not ids
|
|
or any(not isinstance(item, str) or not item for item in ids)
|
|
or len(ids) != len(set(ids))
|
|
):
|
|
return False
|
|
if (
|
|
forward.get("owner_feature_id") != reverse.get("owner_feature_id")
|
|
or {forward.get("output_role"), reverse.get("output_role")} != {"extrude.start", "extrude.end"}
|
|
or (forward.get("selector_intent") or {}).get("disambiguation", {}).get("source_profile_entity_ids")
|
|
!= (reverse.get("selector_intent") or {}).get("disambiguation", {}).get("source_profile_entity_ids")
|
|
):
|
|
return False
|
|
params = producer.get("params") or {}
|
|
if (
|
|
producer.get("atomic_id") != "extrude_add_two_sided"
|
|
or params.get("result_mode") != "new_body"
|
|
or (params.get("end_condition") or {}).get("type") != "blind"
|
|
or (params.get("reverse_end_condition") or {}).get("type") != "blind"
|
|
or params.get("draft") is not None
|
|
):
|
|
return False
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
profile = sketch.get("profile") or {}
|
|
source_sketch_id = sketch.get("source_sketch_id")
|
|
ids = set((forward.get("selector_intent") or {}).get("disambiguation", {}).get("source_profile_entity_ids") or ())
|
|
direct_ids = {str(profile.get("source_entity_id"))} if profile.get("type") == "circle" and profile.get("source_entity_id") else set()
|
|
for contour in profile.get("contours") or ():
|
|
for segment in (contour or {}).get("segments") or ():
|
|
entity_id = segment.get("source_entity_id") if isinstance(segment, dict) else None
|
|
if isinstance(entity_id, str) and entity_id:
|
|
direct_ids.add(entity_id)
|
|
return isinstance(source_sketch_id, str) and bool(direct_ids) and ids == direct_ids
|
|
|
|
|
|
def is_initial_two_sided_circle_shell_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the direct symmetric-circle cap contract for shell removal.
|
|
|
|
The two far prism caps are distinct exact builder results. This is not a
|
|
general two-sided CAP_FACE capability: the source query must name the
|
|
sole original circle and the consumer may use the role only immediately
|
|
as a shell removal face.
|
|
"""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("output_role") not in {"extrude.start", "extrude.end"}
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role_source") is not None
|
|
or any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id"))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_FACE"
|
|
or intent.get("evidence") != "operation_role"
|
|
or intent.get("output_role") != selector.get("output_role")
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
if not isinstance(policy, dict) or policy.get("multiplicity") != "one" or set(policy.get("allowed") or ()) != {"boundary"}:
|
|
return False
|
|
params = producer.get("params") or {}
|
|
if (
|
|
producer.get("atomic_id") != "extrude_add_two_sided"
|
|
or params.get("result_mode") != "new_body"
|
|
or (params.get("end_condition") or {}).get("type") != "blind"
|
|
or (params.get("reverse_end_condition") or {}).get("type") != "blind"
|
|
or params.get("draft") is not None
|
|
):
|
|
return False
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
profile = sketch.get("profile") or {}
|
|
source_sketch_id = sketch.get("source_sketch_id")
|
|
source_entity_id = profile.get("source_entity_id")
|
|
source_entity = intent.get("source_entity")
|
|
return (
|
|
profile.get("type") == "circle"
|
|
and isinstance(source_sketch_id, str)
|
|
and isinstance(source_entity_id, str)
|
|
and source_entity == {
|
|
"sketch_id": source_sketch_id,
|
|
"entity_id": source_entity_id,
|
|
}
|
|
)
|
|
|
|
|
|
def is_initial_direct_loft_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the exact endpoint-role contract for an initial direct loft."""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
params = producer.get("params") or {}
|
|
profile_sources = params.get("cap_output_profile_sources")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("output_role") not in {"loft.start", "loft.end"}
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role_source") is not None
|
|
or any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id"))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_FACE"
|
|
or intent.get("evidence") != "operation_role"
|
|
or intent.get("output_role") != selector.get("output_role")
|
|
or producer.get("atomic_id") != "loft_add"
|
|
or params.get("initial_output_roles") is not True
|
|
or not isinstance(profile_sources, list)
|
|
or len(profile_sources) != 2
|
|
or len(set(profile_sources)) != 2
|
|
):
|
|
return False
|
|
profile_sketch_ids = params.get("profile_sketch_ids")
|
|
if (
|
|
not isinstance(profile_sketch_ids, list)
|
|
or len(profile_sketch_ids) != 2
|
|
or any(not isinstance(sketch_id, str) for sketch_id in profile_sketch_ids)
|
|
or [
|
|
(sketches.get(sketch_id) or {}).get("source_sketch_id")
|
|
for sketch_id in profile_sketch_ids
|
|
] != profile_sources
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
disambiguation = intent.get("disambiguation")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or set(policy.get("allowed") or ()) != {"boundary"}
|
|
or not isinstance(disambiguation, dict)
|
|
or disambiguation.get("type") != "loft_profile_source"
|
|
):
|
|
return False
|
|
source = disambiguation.get("source_sketch_id")
|
|
if source not in profile_sources:
|
|
return False
|
|
return selector["output_role"] == ("loft.start" if profile_sources.index(source) == 0 else "loft.end")
|
|
|
|
|
|
def is_initial_direct_sweep_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate one direct PipeShell cap selected by its exact source pair.
|
|
|
|
A sweep cap has two source qualifications: the profile edge and the path
|
|
endpoint. ``FirstShape``/``LastShape`` prove the final face, while the
|
|
CDSL contract keeps both FeatureScript anchors so ``isStart`` never
|
|
degenerates into an arbitrary cap-role choice.
|
|
"""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
params = producer.get("params") or {}
|
|
contract = params.get("cap_output_contract")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("output_role") not in {"sweep.start", "sweep.end"}
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role_source") is not None
|
|
or any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id"))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_FACE"
|
|
or intent.get("evidence") != "operation_role"
|
|
or intent.get("output_role") != selector.get("output_role")
|
|
or producer.get("atomic_id") != "sweep_add"
|
|
or params.get("result_mode") != "new_body"
|
|
or params.get("initial_output_roles") is not True
|
|
or not isinstance(contract, dict)
|
|
):
|
|
return False
|
|
required = ("profile_source", "profile_entity", "path_source", "path_entity", "path_reversed")
|
|
if (
|
|
any(not isinstance(contract.get(name), str) or not contract[name] for name in required[:-1])
|
|
or not isinstance(contract.get("path_reversed"), bool)
|
|
):
|
|
return False
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
if sketch.get("source_sketch_id") != contract["profile_source"]:
|
|
return False
|
|
profile = sketch.get("profile") or {}
|
|
contours = profile.get("contours")
|
|
if not (
|
|
profile.get("type") == "circle"
|
|
or profile.get("type") == "analytic_contours"
|
|
and isinstance(contours, list)
|
|
and len(contours) == 1
|
|
and bool((contours[0] or {}).get("closed"))
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
disambiguation = intent.get("disambiguation")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or set(policy.get("allowed") or ()) != {"boundary"}
|
|
or not isinstance(disambiguation, dict)
|
|
or disambiguation.get("type") != "sweep_profile_path_endpoint"
|
|
or any(disambiguation.get(name) != contract[name] for name in required)
|
|
or disambiguation.get("path_endpoint") not in {"start", "end"}
|
|
):
|
|
return False
|
|
endpoint = disambiguation["path_endpoint"]
|
|
role_endpoint = endpoint if not contract["path_reversed"] else ("end" if endpoint == "start" else "start")
|
|
return selector["output_role"] == f"sweep.{role_endpoint}"
|
|
|
|
|
|
def is_initial_direct_sweep_cap_edge(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the singleton profile-edge form of a direct PipeShell cap.
|
|
|
|
This does not claim PipeShell provides generic per-edge history. It is
|
|
limited to a direct circular profile, whose one source edge and one cap
|
|
boundary edge are both independently cardinality-checked by the adapter.
|
|
"""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
params = producer.get("params") or {}
|
|
contract = params.get("cap_output_contract")
|
|
if (
|
|
selector.get("kind") != "edge"
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id", "output_role", "output_role_source",
|
|
))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_EDGE"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or producer.get("atomic_id") != "sweep_add"
|
|
or params.get("result_mode") != "new_body"
|
|
or params.get("initial_output_roles") is not True
|
|
or not isinstance(contract, dict)
|
|
):
|
|
return False
|
|
required = ("profile_source", "profile_entity", "path_source", "path_entity", "path_reversed")
|
|
if (
|
|
any(not isinstance(contract.get(name), str) or not contract[name] for name in required[:-1])
|
|
or not isinstance(contract.get("path_reversed"), bool)
|
|
):
|
|
return False
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
profile = sketch.get("profile") or {}
|
|
if (
|
|
sketch.get("source_sketch_id") != contract["profile_source"]
|
|
or profile.get("type") != "circle"
|
|
or profile.get("source_entity_id") != contract["profile_entity"]
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
source_entity = intent.get("source_entity")
|
|
disambiguation = intent.get("disambiguation")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "one"
|
|
or set(policy.get("allowed") or ()) != {"boundary"}
|
|
or source_entity != {"sketch_id": contract["profile_source"], "entity_id": contract["profile_entity"]}
|
|
or not isinstance(disambiguation, dict)
|
|
or disambiguation.get("type") != "sweep_profile_path_endpoint"
|
|
or any(disambiguation.get(name) != contract[name] for name in required)
|
|
or disambiguation.get("path_endpoint") not in {"start", "end"}
|
|
):
|
|
return False
|
|
endpoint = disambiguation["path_endpoint"]
|
|
role_endpoint = endpoint if not contract["path_reversed"] else ("end" if endpoint == "start" else "start")
|
|
return intent.get("lineage_role") == f"sweep.{role_endpoint}"
|
|
|
|
|
|
def is_initial_direct_sweep_swept_face(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the direct analytic-profile PipeShell `Generated(edge)` form."""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
params = producer.get("params") or {}
|
|
contract = params.get("swept_face_contract")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id", "output_role", "output_role_source",
|
|
))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "SWEPT_FACE"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or producer.get("atomic_id") != "sweep_add"
|
|
or params.get("result_mode") != "new_body"
|
|
or params.get("initial_output_roles") is not True
|
|
or not isinstance(contract, dict)
|
|
):
|
|
return False
|
|
required = ("profile_source", "profile_entities", "path_source", "path_entity", "path_reversed")
|
|
if (
|
|
any(not isinstance(contract.get(name), str) or not contract[name] for name in ("profile_source", "path_source", "path_entity"))
|
|
or not isinstance(contract.get("profile_entities"), list)
|
|
or not contract["profile_entities"]
|
|
or any(not isinstance(entity, str) or not entity for entity in contract["profile_entities"])
|
|
or len(set(contract["profile_entities"])) != len(contract["profile_entities"])
|
|
or not isinstance(contract.get("path_reversed"), bool)
|
|
):
|
|
return False
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
profile_entities = _direct_sweep_profile_entities(sketch)
|
|
if sketch.get("source_sketch_id") != contract["profile_source"] or profile_entities != contract["profile_entities"]:
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
disambiguation = intent.get("disambiguation")
|
|
return (
|
|
isinstance(policy, dict)
|
|
and policy.get("multiplicity") == "one"
|
|
and set(policy.get("allowed") or ()) == {"boundary"}
|
|
and isinstance(intent.get("source_entity"), dict)
|
|
and intent["source_entity"].get("sketch_id") == contract["profile_source"]
|
|
and intent["source_entity"].get("entity_id") in profile_entities
|
|
and isinstance(disambiguation, dict)
|
|
and disambiguation.get("type") == "sweep_profile_path"
|
|
and all(disambiguation.get(name) == contract[name] for name in required)
|
|
)
|
|
|
|
|
|
def is_initial_direct_sweep_swept_edge(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the direct profile-vertex PipeShell `Generated(vertex)` form."""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
params = producer.get("params") or {}
|
|
contract = params.get("swept_edge_contract")
|
|
if (
|
|
selector.get("kind") != "edge"
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or any(selector.get(key) is not None for key in (
|
|
"stable_id", "snapshot_id", "geometry", "binding_feature_id", "output_role", "output_role_source",
|
|
))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "SWEPT_EDGE"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or producer.get("atomic_id") != "sweep_add"
|
|
or params.get("result_mode") != "new_body"
|
|
or params.get("initial_output_roles") is not True
|
|
or not isinstance(contract, dict)
|
|
):
|
|
return False
|
|
required = ("profile_source", "profile_entities", "path_source", "path_entity", "path_reversed")
|
|
if (
|
|
any(not isinstance(contract.get(name), str) or not contract[name] for name in ("profile_source", "path_source", "path_entity"))
|
|
or contract.get("profile_source") == contract.get("path_source")
|
|
or not isinstance(contract.get("profile_entities"), list)
|
|
or len(contract["profile_entities"]) < 2
|
|
or any(not isinstance(entity, str) or not entity for entity in contract["profile_entities"])
|
|
or len(set(contract["profile_entities"])) != len(contract["profile_entities"])
|
|
or not isinstance(contract.get("path_reversed"), bool)
|
|
):
|
|
return False
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
profile_entities = _direct_sweep_profile_entities(sketch)
|
|
if sketch.get("source_sketch_id") != contract["profile_source"] or profile_entities != contract["profile_entities"]:
|
|
return False
|
|
source_entities = intent.get("source_entities")
|
|
if not isinstance(source_entities, list) or len(source_entities) != 2:
|
|
return False
|
|
vertex_entities = tuple(sorted(
|
|
source.get("entity_id") for source in source_entities
|
|
if isinstance(source, dict) and source.get("sketch_id") == contract["profile_source"]
|
|
and isinstance(source.get("entity_id"), str)
|
|
))
|
|
policy = intent.get("derivation_policy")
|
|
disambiguation = intent.get("disambiguation")
|
|
return (
|
|
len(vertex_entities) == 2
|
|
and len(set(vertex_entities)) == 2
|
|
and vertex_entities in _direct_sweep_profile_vertex_entity_pairs(sketch)
|
|
and isinstance(policy, dict)
|
|
and policy.get("multiplicity") == "one"
|
|
and set(policy.get("allowed") or ()) == {"boundary"}
|
|
and isinstance(disambiguation, dict)
|
|
and disambiguation.get("type") == "sweep_profile_vertex_path"
|
|
and all(disambiguation.get(name) == contract[name] for name in required)
|
|
and disambiguation.get("profile_vertex_entities") == list(vertex_entities)
|
|
)
|
|
|
|
|
|
def _direct_sweep_profile_entities(sketch: dict[str, Any]) -> list[str] | None:
|
|
"""Return a direct closed profile's complete, unique source edge set."""
|
|
profile = sketch.get("profile") or {}
|
|
if profile.get("type") == "circle":
|
|
entity = profile.get("source_entity_id")
|
|
return [entity] if isinstance(entity, str) and entity else None
|
|
contours = profile.get("contours") if profile.get("type") == "analytic_contours" else None
|
|
if not isinstance(contours, list) or len(contours) != 1:
|
|
return None
|
|
contour = contours[0] or {}
|
|
segments = contour.get("segments") if isinstance(contour, dict) else None
|
|
if not contour.get("closed") or not isinstance(segments, list) or not segments:
|
|
return None
|
|
entities = [segment.get("source_entity_id") for segment in segments if isinstance(segment, dict)]
|
|
if len(entities) != len(segments) or any(not isinstance(entity, str) or not entity for entity in entities):
|
|
return None
|
|
return entities if len(set(entities)) == len(entities) else None
|
|
|
|
|
|
def _direct_sweep_profile_vertex_entity_pairs(sketch: dict[str, Any]) -> set[tuple[str, str]]:
|
|
"""Return only the exact adjacent direct source-edge pairs of one contour."""
|
|
entities = _direct_sweep_profile_entities(sketch)
|
|
if entities is None or len(entities) < 2:
|
|
return set()
|
|
return {
|
|
tuple(sorted((entities[index], entities[(index + 1) % len(entities)])))
|
|
for index in range(len(entities))
|
|
}
|
|
|
|
|
|
def is_planar_imprint_extrude_cap_output_role(
|
|
selector: dict[str, Any],
|
|
producer: dict[str, Any] | None,
|
|
sketches: dict[str, dict[str, Any]],
|
|
) -> bool:
|
|
"""Validate the narrow all-fragment CAP_FACE contract for IMPRINT prisms.
|
|
|
|
A CAP query over an IMPRINT result is only executable when its source
|
|
disambiguation names the complete profile source set. The runtime still
|
|
verifies every generated cap fragment through fresh builder history.
|
|
"""
|
|
if not isinstance(selector, dict) or not isinstance(producer, dict):
|
|
return False
|
|
intent = selector.get("selector_intent")
|
|
if (
|
|
selector.get("kind") != "face"
|
|
or selector.get("output_role") not in {"extrude.start", "extrude.end"}
|
|
or selector.get("source") != "runtime_snapshot"
|
|
or selector.get("output_role_source") is not None
|
|
or any(selector.get(key) is not None for key in ("stable_id", "snapshot_id", "geometry", "binding_feature_id"))
|
|
or not isinstance(intent, dict)
|
|
or intent.get("query_family") != "CAP_FACE"
|
|
or intent.get("evidence") != "kernel_history"
|
|
or intent.get("output_role") != selector.get("output_role")
|
|
):
|
|
return False
|
|
policy = intent.get("derivation_policy")
|
|
if (
|
|
not isinstance(policy, dict)
|
|
or policy.get("multiplicity") != "all_fragments"
|
|
or set(policy.get("allowed") or ()) != {"boundary", "fragment"}
|
|
):
|
|
return False
|
|
params = producer.get("params") or {}
|
|
if (
|
|
producer.get("atomic_id") not in {"extrude_add_blind", "extrude_cut_blind"}
|
|
or params.get("result_mode") != "new_body"
|
|
or (params.get("end_condition") or {}).get("type") != "blind"
|
|
or params.get("draft") is not None
|
|
):
|
|
return False
|
|
sketch = sketches.get(str(producer.get("sketch_id") or "")) or {}
|
|
profile = sketch.get("profile") or {}
|
|
if profile.get("type") != "planar_imprint":
|
|
return False
|
|
expected_ids = sorted({
|
|
str(entry.get("id"))
|
|
for entry in profile.get("source_entities") or ()
|
|
if isinstance(entry, dict) and isinstance(entry.get("id"), str) and entry.get("id")
|
|
})
|
|
disambiguation = intent.get("disambiguation")
|
|
actual_ids = sorted({
|
|
str(value)
|
|
for value in (disambiguation or {}).get("source_entity_ids") or ()
|
|
if isinstance(value, str) and value
|
|
})
|
|
return (
|
|
bool(expected_ids)
|
|
and isinstance(disambiguation, dict)
|
|
and disambiguation.get("type") == "complete_imprint_profile_source_set"
|
|
and actual_ids == expected_ids
|
|
and len(actual_ids) == len((disambiguation or {}).get("source_entity_ids") or ())
|
|
)
|