| """Deterministic AOOE controller -> canonical transition compiler. |
| |
| This module is a runtime representation boundary. It may bind facts already |
| present in an ontology-derived governance projection or normalize equivalent |
| field/action spellings. It must not repair a semantically wrong controller |
| choice (for example, inspect -> abstain or add_metadata -> link_existing). |
| """ |
| from __future__ import annotations |
|
|
| import copy |
| from typing import Any |
|
|
| CANONICAL_ACTIONS = { |
| "inspect", "create_entity", "set_fields", "set_publication", |
| "delete_entity", "abstain", |
| } |
| RELATION_FIELDS = {"place", "region"} |
|
|
|
|
| def _projection(payload: dict[str, Any]) -> dict[str, Any] | None: |
| value = payload.get("governance_projection") |
| return value if isinstance(value, dict) else None |
|
|
|
|
| def _target(projection: dict[str, Any]) -> str | None: |
| value = projection.get("target_entity", projection.get("targetEntity")) |
| return value if isinstance(value, str) and value else None |
|
|
|
|
| def _operation(projection: dict[str, Any]) -> str: |
| return str(projection.get("operation_class", projection.get("operationFamily", "")) or "") |
|
|
|
|
| def _required_fields(projection: dict[str, Any]) -> list[str]: |
| value = projection.get("required_fields", projection.get("requiredFields", [])) |
| return [str(v) for v in value] if isinstance(value, list) else [] |
|
|
|
|
| def _projected_entity(payload: dict[str, Any], target: str | None) -> dict[str, Any]: |
| state = payload.get("projected_state") |
| if not isinstance(state, dict): |
| return {} |
| entities = state.get("entities") |
| if target and isinstance(entities, dict) and isinstance(entities.get(target), dict): |
| return entities[target] |
| entity = state.get("entity") |
| return entity if isinstance(entity, dict) else {} |
|
|
|
|
| def _escalates(action: dict[str, Any]) -> bool: |
| if action.get("escalate") is True or action.get("disposition") == "escalate": |
| return True |
| planned = action.get("plannedTransitions") |
| return isinstance(planned, list) and any("escalat" in str(v).lower() for v in planned) |
|
|
|
|
| def _normalize_fields(fields: dict[str, Any]) -> dict[str, Any]: |
| out = copy.deepcopy(fields) |
| aliases = { |
| "entity_kind": "entityKind", |
| "canonical_url": "canonicalUrl", |
| } |
| for old, new in aliases.items(): |
| if old in out and new not in out: |
| out[new] = out.pop(old) |
| return out |
|
|
|
|
| def _candidate_fields(action: dict[str, Any]) -> dict[str, Any] | None: |
| fields = action.get("fields") |
| if isinstance(fields, dict): |
| return _normalize_fields(fields) |
| details = action.get("details") |
| if isinstance(details, dict): |
| nested = details.get("fields") |
| if isinstance(nested, dict): |
| return _normalize_fields(nested) |
| return None |
|
|
|
|
| def _explicit_relationship_fields(action: dict[str, Any]) -> dict[str, Any] | None: |
| """Extract only relationship predicates the controller explicitly emitted.""" |
| details = action.get("details") |
| if not isinstance(details, dict): |
| return None |
| fields = { |
| key: copy.deepcopy(details[key]) |
| for key in RELATION_FIELDS |
| if key in details and isinstance(details[key], str) and details[key] |
| } |
| return fields or None |
|
|
|
|
| def _explicit_reference_ids(action: dict[str, Any]) -> list[str]: |
| """Collect reference IDs explicitly carried by a relationship decision. |
| |
| This normalizes carrier names only. It does not infer a missing predicate or |
| add references from the governance projection. |
| """ |
| refs: list[str] = [] |
| targets = action.get("targets") |
| if isinstance(targets, list): |
| refs.extend(v for v in targets if isinstance(v, str) and v) |
| details = action.get("details") |
| if isinstance(details, dict): |
| for key in ("references", "relationTargets", "ids"): |
| values = details.get(key) |
| if isinstance(values, list): |
| refs.extend(v for v in values if isinstance(v, str) and v) |
| seen: set[str] = set() |
| return [v for v in refs if not (v in seen or seen.add(v))] |
|
|
|
|
| def _record_kind(action: dict[str, Any]) -> str | None: |
| for key in ("entityKind", "entity_kind", "recordKind", "record_kind"): |
| value = action.get(key) |
| if isinstance(value, str) and value: |
| return value |
| return None |
|
|
|
|
| def compile_governed_action(action: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]: |
| """Lower one governed controller action without inventing semantic intent. |
| |
| If ``payload`` is not an AOOE/governance projection, the action is returned |
| unchanged. The compiler only uses the supplied projection/projected state. |
| """ |
| projection = _projection(payload) |
| if projection is None or not isinstance(action, dict): |
| return action |
|
|
| out = copy.deepcopy(action) |
| kind = str(out.get("action") or "") |
| target = _target(projection) |
|
|
| if kind == "inspect": |
| if target and not isinstance(out.get("entity_id"), str): |
| out["entity_id"] = target |
| findings = out.get("findings") |
| if not isinstance(findings, list): |
| required = _required_fields(projection) |
| entity = _projected_entity(payload, target) |
| if required: |
| out["findings"] = [ |
| field for field in required |
| if field not in entity or entity.get(field) in (None, "", [], {}) |
| ] |
| else: |
| aliases = out.get("validationTargets") |
| if isinstance(aliases, list): |
| out["findings"] = [str(v) for v in aliases] |
| return out |
|
|
| if kind in {"create_entity", "create-place", "create_place", "create-trail", "create_trail"}: |
| expected_kind = str(projection.get("expectedRecordKind") or "") |
| explicit_family = "trail" if "trail" in kind else ("place" if "place" in kind else "") |
| supplied_kind = _record_kind(out) |
| if explicit_family and expected_kind and explicit_family != expected_kind.lower(): |
| return out |
| if supplied_kind and expected_kind and supplied_kind.lower() != expected_kind.lower(): |
| return out |
|
|
| fields = _candidate_fields(out) |
| if fields is not None: |
| record_kind = supplied_kind or expected_kind |
| if record_kind and "entityKind" not in fields: |
| fields["entityKind"] = record_kind |
|
|
| allowed = projection.get("allowed_new_entity", projection.get("allowedNewEntity")) |
| if kind == "create_entity": |
| if fields is not None: |
| out["fields"] = fields |
| if isinstance(allowed, str) and allowed: |
| out.setdefault("entity_id", allowed) |
| return out |
|
|
| if fields is None: |
| return out |
| if not isinstance(allowed, str) or not allowed: |
| return out |
| return {"action": "create_entity", "entity_id": allowed, "fields": fields} |
|
|
| if kind == "set_fields": |
| if target and not isinstance(out.get("entity_id"), str): |
| out["entity_id"] = target |
| fields = _candidate_fields(out) |
| if fields is not None: |
| out["fields"] = fields |
| return out |
|
|
| if kind in {"add_metadata", "apply_governance", "update_governance"}: |
| fields = _candidate_fields(out) |
| if target and fields is not None: |
| return {"action": "set_fields", "entity_id": target, "fields": fields} |
| return out |
|
|
| if kind in {"publish", "set_publication"}: |
| transition = projection.get("allowed_publication_transition", projection.get("allowedPublicationTransition")) |
| status = out.get("publication_status") |
| if not isinstance(status, str) and isinstance(transition, list) and len(transition) == 2: |
| status = transition[1] |
| if target and isinstance(status, str): |
| return {"action": "set_publication", "entity_id": target, "publication_status": status} |
| return out |
|
|
| if kind == "abstain": |
| requires = projection.get("requires_escalation_if_abstaining", projection.get("requiresEscalation")) is True |
| if _escalates(out) or requires: |
| out["escalate"] = True |
| return out |
|
|
| if kind in {"link_existing", "link-existing", "link_entities"}: |
| fields = _candidate_fields(out) |
| if fields is None: |
| fields = _explicit_relationship_fields(out) |
| if target and fields is not None: |
| return {"action": "set_fields", "entity_id": target, "fields": fields} |
| refs = _explicit_reference_ids(out) |
| if refs: |
| out["targets"] = refs |
| return out |
|
|
| return out |
|
|
