#@title 7. UNIVERSAL VALIDATION / RANKING / ENGINE v5.6 — robust source-bounded non-generative integrity judge # Complete replacement for Cell 7. # Compatible with Cell 5 v5, Cell 5A v4.3, Cell 6 v6.1. # Accepts arbitrary Unicode text safely; linguistic routing quality still depends # on the loaded spaCy model and available annotations. # Run Cell 7A and Cell 7B again after this cell. import json import re import time import unicodedata from collections import Counter from dataclasses import asdict from difflib import SequenceMatcher from pathlib import Path EXPECTED_CELL7_RULE_COUNT = 42 _REQUIRED_CELL7 = [ "STATE", "CFG", "RULES", "Analyzer", "Generator", "Scanner", "Candidate", "SentenceResult", "DocumentResult", "PROFILES", "protect", "optional", "nlp", "sub", "sent_split", "surface_key", "realization_issues", ] _missing_cell7 = [name for name in _REQUIRED_CELL7 if name not in globals()] if _missing_cell7: raise RuntimeError( "Cell 7 cannot run. Missing: " + ", ".join(_missing_cell7) + ". Run Cells 1–6 v6.1 in order first." ) if len(RULES) != EXPECTED_CELL7_RULE_COUNT: raise RuntimeError( f"Cell 7 expected {EXPECTED_CELL7_RULE_COUNT} rules but found {len(RULES)}." ) # Backward compatibility for older tests/helpers. _surface_key = surface_key NEG = {"not", "no", "never", "without", "neither", "nor"} HEDGE = { "may", "might", "could", "can", "possibly", "potentially", "suggest", "suggests", "indicate", "indicates", "likely", "approximately", "about", "must", "should", "would", "will", } def _ensure_state(): if not isinstance(STATE, dict): raise RuntimeError("STATE must be a dictionary.") STATE.setdefault("warnings", []) STATE.setdefault("start", None) return STATE def sr(a, b): return SequenceMatcher( None, re.sub(r"\s+", " ", str(a).lower()).strip(), re.sub(r"\s+", " ", str(b).lower()).strip(), ).ratio() def token_change(a, b): # Unicode-aware lexical change. Punctuation and dash variants are # comparison boundaries, while the visible candidate text remains intact. A = _unicode_tokens(a) if "_unicode_tokens" in globals() else re.findall(r"\w+", str(a).casefold(), flags=re.UNICODE) B = _unicode_tokens(b) if "_unicode_tokens" in globals() else re.findall(r"\w+", str(b).casefold(), flags=re.UNICODE) return 1 - SequenceMatcher(None, A, B).ratio() def nli_pair(a, b): _ensure_state() bundle = optional("nli") if not bundle: return None try: import torch tok, model = bundle x = tok(a, b, return_tensors="pt", truncation=True, max_length=512) dev = next(model.parameters()).device x = {k: v.to(dev) for k, v in x.items()} with torch.no_grad(): prob = torch.softmax(model(**x).logits[0], -1).cpu().numpy() labels = { str(model.config.id2label[i]).lower(): float(prob[i]) for i in range(len(prob)) } return { "entailment": max([v for k, v in labels.items() if "entail" in k] or [0]), "contradiction": max([v for k, v in labels.items() if "contrad" in k] or [0]), } except Exception as exc: STATE["warnings"].append(f"NLI scoring failed: {exc}") return None def stsim(a, b): _ensure_state() model = optional("sts") if not model: return None try: from sentence_transformers.util import cos_sim emb = model.encode([a, b], convert_to_tensor=True, normalize_embeddings=True) return float(cos_sim(emb[0], emb[1]).item()) except Exception as exc: STATE["warnings"].append(f"STS scoring failed: {exc}") return None # ===================================================================== # RELATION / INVENTORY HELPERS # ===================================================================== def _safe_doc(text): try: return nlp()(str(text or "")) except Exception: return None def _neg_inventory(text, doc=None): """Compare polarity markers by lexical class, not dependency owner.""" lo = str(text or "").lower() return Counter( token for token in ["not", "no", "never", "without", "neither", "nor"] if re.search(r"(?.+?)\s+but\s+to\s+(?P.+?)[.]?$", original, ) if not match: return False x = surface_key(match["x"]) y = surface_key(match["y"]) z = surface_key(candidate) return ( x in z and y in z and ( "rather than" in candidate.lower() or ("does not" in candidate.lower() and "instead" in candidate.lower()) ) ) def _unicode_normalize(text): """Canonical Unicode normalization for comparison, never for output.""" value = unicodedata.normalize("NFKC", str(text or "")) translations = str.maketrans({ "–": "-", "—": "-", "−": "-", "‑": "-", "’": "'", "‘": "'", "“": '"', "”": '"', "\u00a0": " ", }) return re.sub(r"\s+", " ", value.translate(translations)).strip().casefold() def _unicode_tokens(text): """ Tokenize letters/numbers across Unicode scripts without ASCII loss. Punctuation, including every dash/apostrophe form, is treated as a comparison boundary. This makes these equivalent for inventory checks: re-examination == re examination == re–examination matched-data == matched data The original candidate text is never modified. """ value = _unicode_normalize(text) output = [] current = [] for char in value: category = unicodedata.category(char) if category[0] in {"L", "N", "M"}: current.append(char) elif current: token = "".join(current) if token: output.append(token) current = [] if current: token = "".join(current) if token: output.append(token) return output def _noun_chunk_for_token(doc, token): try: chunks = [chunk for chunk in doc.noun_chunks if chunk.start <= token.i < chunk.end] except Exception: chunks = [] return min(chunks, key=lambda chunk: chunk.end - chunk.start) if chunks else None def _clean_member_text(text): value = re.sub(r"\s+", " ", str(text or "")).strip(" \t\r\n,;:.()[]{}") value = re.sub(r"^(?:and|or|nor)\s+", "", value, flags=re.I) return value.strip() def _dependency_enumeration_groups(doc): """Recover coordinated members from dependency heads/noun chunks.""" if doc is None or not doc.has_annotation("DEP"): return [] groups = [] seen = set() for token in doc: conjuncts = [child for child in token.children if child.dep_ == "conj"] if not conjuncts: continue members = [token] + sorted(conjuncts, key=lambda item: item.i) # Include chained conjuncts. cursor = 0 while cursor < len(members): for child in members[cursor].children: if child.dep_ == "conj" and child not in members: members.append(child) cursor += 1 members = sorted(set(members), key=lambda item: item.i) texts = [] for member in members: chunk = _noun_chunk_for_token(doc, member) if chunk is not None: text = chunk.text else: # Conservative local subtree: exclude other conjunct branches. blocked = {item.i for other in members if other is not member for item in other.subtree} indices = [ item.i for item in member.subtree if item.i not in blocked and item.dep_ not in {"cc", "punct"} ] if indices: text = doc[min(indices):max(indices) + 1].text else: text = member.text text = _clean_member_text(text) if text: texts.append(text) texts = list(dict.fromkeys(texts)) key = tuple(_unicode_normalize(item) for item in texts) if len(texts) >= 3 and key not in seen: seen.add(key) groups.append(texts) return groups def _analysis_enumeration_groups(analysis): """Accept analyzer groups only when each item is a plausible member.""" groups = [] for group in list(getattr(analysis, "enumerations", []) or []): cleaned = [_clean_member_text(item) for item in group] cleaned = [item for item in cleaned if item] # Reject boundary-corrupted pseudo-members containing multiple commas. if len(cleaned) >= 3 and all(item.count(",") <= 1 for item in cleaned): groups.append(cleaned) return groups def _top_level_segments(text): """Split on top-level commas while respecting brackets and Unicode text.""" segments = [] current = [] stack = [] pairs = {"(": ")", "[": "]", "{": "}"} closers = set(pairs.values()) for char in str(text or ""): if char in pairs: stack.append(pairs[char]) elif char in closers and stack and stack[-1] == char: stack.pop() if char == "," and not stack: segments.append("".join(current).strip()) current = [] else: current.append(char) if current: segments.append("".join(current).strip()) return segments def _surface_list_groups(text): """Conservative fallback; never cuts through hyphenated compounds.""" value = str(text or "").strip().rstrip(".?!") segments = _top_level_segments(value) if len(segments) < 3: return [] # Locate the final coordinator in the final segment. final = segments[-1] match = re.search(r"\s+(?:and|or|nor)\s+(.+)$", final, re.I | re.S) if not match: return [] last_member = _clean_member_text(match.group(1)) pre_final = _clean_member_text(final[:match.start()]) # Find the most plausible list start. Prefer explicit introducers. introducer = re.compile( r"\b(?:using|including|involving|such as|comprising|consisting of|covers?|" r"evaluated|assessed|measured|reviewed|identified|pre-register(?:ed)?|preregister(?:ed)?)\s+", re.I, ) first_index = 0 first_text = segments[0] matches = list(introducer.finditer(first_text)) if matches: first_member = _clean_member_text(first_text[matches[-1].end():]) else: # Without an introducer, avoid guessing a list start from the full clause. return [] middle = [_clean_member_text(item) for item in segments[1:-1]] members = [first_member] + middle if pre_final: members.append(pre_final) if last_member: members.append(last_member) members = [item for item in members if item] return [members] if len(members) >= 3 else [] def _plausible_enumeration_member(text): """ Reject boundary-corrupted pseudo-members without weakening true omission checks. A valid member may contain Unicode letters, numbers, spaces, apostrophes, and dash variants, but should not contain clause punctuation, a full sentence lead-in, or an excessive token span. """ value = _clean_member_text(text) tokens = _unicode_tokens(value) if not tokens or len(tokens) > 12: return False if any(mark in value for mark in [",", ";", ":", "."]): return False if re.match( r"(?is)^(?:we|i|they|he|she|it|the\s+(?:study|paper|article|report|" r"system|framework|team|model))\s+" r"(?:propose|proposes|introduce|introduces|present|presents|" r"evaluate|evaluates|assess|assesses|pre-register|preregister)\b", value, ): return False return True def _enumeration_groups(analysis, source_text, source_doc): """ Build only high-confidence list groups. Sources are tried independently instead of accepting the first nonempty extractor. Candidate groups are filtered for plausible member boundaries, verified against the source, de-duplicated, and ranked by compactness. """ candidate_groups = [] candidate_groups.extend(_dependency_enumeration_groups(source_doc)) candidate_groups.extend(_analysis_enumeration_groups(analysis)) candidate_groups.extend(_surface_list_groups(source_text)) source_tokens = _unicode_tokens(source_text) accepted = [] seen = set() for group in candidate_groups: stable_group = [] for raw_item in group: item = _clean_member_text(raw_item) if not _plausible_enumeration_member(item): continue item_tokens = _unicode_tokens(item) width = len(item_tokens) appears_in_source = any( source_tokens[index:index + width] == item_tokens for index in range(len(source_tokens) - width + 1) ) if appears_in_source: stable_group.append(item) # Preserve order and remove duplicate members. unique_group = [] member_seen = set() for item in stable_group: key = tuple(_unicode_tokens(item)) if key and key not in member_seen: member_seen.add(key) unique_group.append(item) normalized = tuple(tuple(_unicode_tokens(item)) for item in unique_group) if len(normalized) >= 3 and normalized not in seen: seen.add(normalized) accepted.append(unique_group) # Remove groups that are strict noisy supersets of a cleaner group. compact = [] token_sets = [ {tuple(_unicode_tokens(item)) for item in group} for group in accepted ] for index, group in enumerate(accepted): current = token_sets[index] noisier = any( other < current for other_index, other in enumerate(token_sets) if other_index != index and len(other) >= 3 ) if not noisier: compact.append(group) return compact def _member_present(member, candidate): """Require normalized member tokens in order, independent of punctuation.""" member_tokens = _unicode_tokens(member) candidate_tokens = _unicode_tokens(candidate) if not member_tokens: return True width = len(member_tokens) return any( candidate_tokens[index:index + width] == member_tokens for index in range(len(candidate_tokens) - width + 1) ) def _relation_signature(text): lo = str(text or "").lower() classes = { "cause": {"because", "therefore", "thus", "consequently", "hence", "as a result", "for this reason", "accordingly"}, "concession": {"although", "though", "despite", "nevertheless", "however", "even so"}, "condition": {"if", "unless", "whether"}, "simultaneous": {"while", "at the same time"}, "purpose": {"so that", "intended to", "designed to"}, "absence": {"without"}, "dependency": {"contingent on", "depend on", "depends on", "depended on"}, } return { key for key, words in classes.items() if any(re.search(r"(?(?:[A-Z]{2,}|[A-Z][A-Z0-9]*\d[A-Z0-9]*)(?:-[A-Z0-9]+)*)\s*\)", flags=re.UNICODE, ) doc = _safe_doc(value) for match in acronym_pattern.finditer(value): acronym = match.group("acronym") prefix = value[:match.start()] # Start after the nearest punctuation or coordinator boundary. boundary = max(prefix.rfind(","), prefix.rfind(";"), prefix.rfind(":")) fragment = prefix[boundary + 1:].strip() coordinator = list(re.finditer(r"(?iu)\b(?:and|or|nor)\s+", fragment)) if coordinator: fragment = fragment[coordinator[-1].end():].strip() # If the fragment still contains a full clause, use the parsed token # ending immediately before '(' to recover its governing noun phrase. expansion = "" if doc is not None: left_tokens = [token for token in doc if token.idx + len(token.text) <= match.start()] if left_tokens: tail = left_tokens[-1] try: chunks = [ chunk for chunk in doc.noun_chunks if chunk.start <= tail.i < chunk.end and not value[chunk.end_char:match.start()].strip() ] except Exception: chunks = [] if chunks: expansion = min(chunks, key=lambda chunk: chunk.end - chunk.start).text.strip() if not expansion: expansion = fragment # Remove only an attested clause lead-in. This is generic: it uses # POS/dependency evidence instead of a list of domain verbs. if doc is not None and expansion == fragment: frag_start = match.start() - len(fragment) frag_tokens = [ token for token in doc if token.idx >= frag_start and token.idx + len(token.text) <= match.start() ] finite_roots = [ token for token in frag_tokens if token.pos_ in {"VERB", "AUX"} and token.dep_ in {"ROOT", "ccomp", "xcomp", "conj"} ] if finite_roots: last_verb = finite_roots[-1] after = value[last_verb.idx + len(last_verb.text):match.start()].strip() if after: expansion = after expansion = re.sub(r"(?iu)^(?:and|or|nor)\s+", "", expansion) expansion = re.sub(r"\s+", " ", expansion).strip(" ,;:") expansion_tokens = tuple(_unicode_tokens(expansion)) # Expansion must be a compact phrase, not a complete clause. if not (1 <= len(expansion_tokens) <= 10): continue if doc is not None: exp_doc = _safe_doc(expansion) if exp_doc is not None and any( token.dep_ == "ROOT" and token.pos_ in {"VERB", "AUX"} and any(child.dep_ in {"nsubj", "csubj"} for child in token.children) for token in exp_doc ): continue pairs.append({ "expansion": expansion, "expansion_tokens": expansion_tokens, "acronym": acronym, "surface": value[max(0, match.start() - len(expansion)):match.end()].strip(), }) return pairs def _acronym_inventory(text): """Count uppercase acronyms only; ordinary title-case words are excluded.""" return Counter( token for token in re.findall( r"(?[A-Z][A-Z0-9-]{1,15})\s+" r"(?:is|are|was|were|equals?|means?)\s+(?P[A-Z][A-Z0-9-]{1,15})(?![\w-])", str(text or ""), ): if match.group("x").casefold() == match.group("y").casefold(): issues.append("REDUNDANT_ACRONYM_COPULAR_TAUTOLOGY") return issues _SOURCE_BOUND_FUNCTION_WORDS = { "a", "an", "the", "and", "or", "nor", "but", "if", "then", "than", "as", "at", "by", "for", "from", "in", "into", "of", "on", "to", "under", "with", "without", "that", "which", "who", "whom", "whose", "this", "these", "those", "it", "its", "is", "are", "was", "were", "be", "been", "being", "do", "does", "did", "have", "has", "had", "not", "no", "so", "thereby", "according", "through", "within", } def _sentence_units(text): """Return normalized sentence strings without changing the candidate.""" value = str(text or "").strip() doc = _safe_doc(value) if doc is not None: try: units = [sent.text.strip() for sent in doc.sents if sent.text.strip()] if units: return units except Exception: pass return [ part.strip() for part in re.split(r"(?<=[.!?])\s+", value) if part.strip() ] def _content_lemmas(text): """Extract content lemmas for comparison; never alter visible output.""" doc = _safe_doc(text) output = [] if doc is not None: for token in doc: if token.is_space or token.is_punct: continue lemma = (token.lemma_ or token.text).casefold().strip() if not lemma or lemma in _SOURCE_BOUND_FUNCTION_WORDS: continue if token.pos_ in {"NOUN", "PROPN", "VERB", "ADJ", "ADV", "NUM"}: output.append(lemma) return output return [ token for token in _unicode_tokens(text) if token not in _SOURCE_BOUND_FUNCTION_WORDS ] def _named_entity_inventory(text): """Count explicit named entities. Empty when a trained NER is unavailable.""" doc = _safe_doc(text) if doc is None: return Counter() try: return Counter( _unicode_normalize(entity.text) for entity in doc.ents if entity.text.strip() ) except Exception: return Counter() def _sentence_start_signatures(text, minimum_width=3, maximum_width=6): """Return bounded lexical prefixes for every sentence. Three content lemmas are sufficient to identify a repeated clause opening such as ``the literature of history encompasses`` even when the next content word differs. Longer prefixes are also retained for diagnostics. """ signatures = [] for sentence in _sentence_units(text): lemmas = _content_lemmas(sentence) if len(lemmas) < minimum_width: continue upper = min(maximum_width, len(lemmas)) signatures.append({ "minimum": tuple(lemmas[:minimum_width]), "prefixes": [tuple(lemmas[:width]) for width in range(minimum_width, upper + 1)], "sentence": sentence, }) return signatures def _repeated_sentence_openings(text, minimum_width=3): """Count repeated sentence openings using a stable minimum prefix.""" records = _sentence_start_signatures(text, minimum_width=minimum_width) return Counter(record["minimum"] for record in records) def _repeated_ngram_inventory(text, minimum=3, maximum=8): """Return repeated lexical n-grams and their counts.""" tokens = _unicode_tokens(text) repeated = Counter() for width in range(minimum, min(maximum, len(tokens)) + 1): counts = Counter(tuple(tokens[i:i + width]) for i in range(len(tokens) - width + 1)) for ngram, count in counts.items(): if count > 1: repeated[ngram] = count return repeated def _source_bounded_integrity_issues(original, revised): """Reject newly added or repeated content in a non-generative rewrite. The test permits grammatical scaffolding and derivational restructuring, but rejects new named entities, duplicated sentences/openings, newly repeated lexical spans, unexplained content-word inflation, and excessive candidate expansion. """ issues = [] source_text = str(original or "") candidate_text = str(revised or "") source_tokens = _unicode_tokens(source_text) candidate_tokens = _unicode_tokens(candidate_text) source_sentences = _sentence_units(source_text) candidate_sentences = _sentence_units(candidate_text) # Exact sentence duplication introduced by the candidate. source_sentence_counts = Counter(_unicode_normalize(x) for x in source_sentences) candidate_sentence_counts = Counter(_unicode_normalize(x) for x in candidate_sentences) duplicated_sentences = { sentence: count for sentence, count in candidate_sentence_counts.items() if count > source_sentence_counts.get(sentence, 0) and count > 1 } if duplicated_sentences: issues.append({ "type": "REPEATED_SENTENCE_INTRODUCED", "items": duplicated_sentences, }) # Repeated sentence openings, the defect visible in the history example. source_starts = _repeated_sentence_openings(source_text, minimum_width=3) candidate_starts = _repeated_sentence_openings(candidate_text, minimum_width=3) repeated_starts = { " ".join(signature): count for signature, count in candidate_starts.items() if count > 1 and count > source_starts.get(signature, 0) } if repeated_starts: issues.append({ "type": "REPEATED_SENTENCE_OPENING_INTRODUCED", "items": repeated_starts, }) # Newly repeated 3+ token spans. source_repeats = _repeated_ngram_inventory(source_text) candidate_repeats = _repeated_ngram_inventory(candidate_text) new_repeats = { " ".join(ngram): count for ngram, count in candidate_repeats.items() if count > source_repeats.get(ngram, 0) } if new_repeats: # Report shortest, most actionable spans only. shortest = sorted(new_repeats.items(), key=lambda item: (len(item[0].split()), item[0]))[:8] issues.append({ "type": "REPEATED_LEXICAL_SPAN_INTRODUCED", "items": dict(shortest), }) # New named entities are a hard factual addition. source_entities = _named_entity_inventory(source_text) candidate_entities = _named_entity_inventory(candidate_text) added_entities = candidate_entities - source_entities if added_entities: issues.append({ "type": "NAMED_ENTITY_INTRODUCED", "items": dict(added_entities), }) # Candidate should not expand substantially. A small allowance supports # relative pronouns and grammatical scaffolding used by deterministic rules. if source_tokens: ratio = len(candidate_tokens) / max(len(source_tokens), 1) allowance = max(5, int(round(len(source_tokens) * 0.18))) if len(candidate_tokens) > len(source_tokens) + allowance and ratio > 1.24: issues.append({ "type": "EXCESSIVE_CONTENT_EXPANSION", "source_tokens": len(source_tokens), "candidate_tokens": len(candidate_tokens), "ratio": round(ratio, 3), }) # Content-lemma counts may change through inflection, but unexplained # multiplicative inflation indicates repetition or added propositions. source_content = Counter(_content_lemmas(source_text)) candidate_content = Counter(_content_lemmas(candidate_text)) inflated = { lemma: {"source": source_content.get(lemma, 0), "candidate": count} for lemma, count in candidate_content.items() if count >= 2 and count > source_content.get(lemma, 0) + 1 and lemma not in _SOURCE_BOUND_FUNCTION_WORDS } if inflated: issues.append({ "type": "CONTENT_LEMMA_REPETITION_INTRODUCED", "items": inflated, }) return issues def _human_style_penalties(original, candidate_text): """Soft human-editor preferences; never override hard safety gates.""" penalties = [] revised = str(candidate_text or "") # Hypothesis labels are findings under evaluation, not evidence sources. if re.search(r"(?iu)\baccording\s+to\s+H\d+[A-Za-z]?\b", revised): penalties.append({"code": "AWKWARD_HYPOTHESIS_ATTRIBUTION", "weight": 0.08}) # Penalize a demonstrative-led sentence introduced by restructuring when # the source did not begin a sentence that way. This is only a tie-breaker. source_demo = len(re.findall(r"(?imu)(?:^|[.!?]\s+)\s*(?:this|that|these|those)\b", str(original or ""))) candidate_demo = len(re.findall(r"(?imu)(?:^|[.!?]\s+)\s*(?:this|that|these|those)\b", revised)) if candidate_demo > source_demo: penalties.append({"code": "NEW_DEMONSTRATIVE_ANTECEDENT", "weight": 0.05}) return penalties def _v5_structural_signature(doc): """Extract vocabulary-independent semantic structure for diagnostics.""" signature = { "roots": [], "subjects": [], "objects": [], "passive": False, "agents": [], "complements": [], } if doc is None or not getattr(doc, "has_annotation", lambda *_: False)("DEP"): return signature for token in doc: if token.dep_ == "ROOT": signature["roots"].append(token.lemma_.casefold()) if token.dep_ in {"nsubj", "nsubjpass", "nsubj:pass", "csubj"}: signature["subjects"].append(_unicode_normalize(sub(token))) if token.dep_ in {"obj", "dobj", "attr", "oprd"}: signature["objects"].append(_unicode_normalize(sub(token))) if token.dep_ in {"nsubjpass", "nsubj:pass", "auxpass", "aux:pass"}: signature["passive"] = True if token.dep_ in {"agent", "obl:agent"}: signature["agents"].append(_unicode_normalize(sub(token))) if token.dep_ in {"ccomp", "xcomp", "advcl", "relcl", "acl"}: signature["complements"].append((token.dep_, token.lemma_.casefold())) return signature def _v5_family_kind(candidate): family = str(getattr(candidate, "family", "") or "") return "universal" if "universal_" in family.casefold() else "specialized" # ===================================================================== # VALIDATOR # ===================================================================== class Validator: def validate(self, original, candidate, analysis, profile, user_terms, mode): _ensure_state() revised = candidate.text failures = [] reasons = [] checks = {} def reject(code, why): if code not in failures: failures.append(code) reasons.append({"code": code, "reason": why}) # ------------------------------------------------------------- # Exact protected inventories # ------------------------------------------------------------- hard_terms = set(getattr(profile, "immutable", set()) or set()) | set(user_terms or set()) for kind, code in [ ("citation", "TERMINOLOGY_REJECTION"), ("quantity", "QUANTITY_REJECTION"), ("url", "TERMINOLOGY_REJECTION"), ("equation", "TERMINOLOGY_REJECTION"), ("code", "TERMINOLOGY_REJECTION"), ]: source_inventory = Counter( item[1].lower() for item in protect(original, hard_terms) if item[0] == kind ) candidate_inventory = Counter( item[1].lower() for item in protect(revised, hard_terms) if item[0] == kind ) checks[kind] = source_inventory == candidate_inventory if source_inventory != candidate_inventory: reject( code, f"{kind} inventory changed: {dict(source_inventory)} -> {dict(candidate_inventory)}", ) missing_terms = [ term for term in hard_terms if re.search(r"(? {dict(candidate_neg)}", ) # ------------------------------------------------------------- # Modality / hedging # ------------------------------------------------------------- source_modals = _modals(source_doc, original) candidate_modals = _modals(candidate_doc, revised) checks["modality"] = (dict(source_modals), dict(candidate_modals)) if source_modals != candidate_modals: reject( "MODALITY_REJECTION", f"Modality/hedging changed: {dict(source_modals)} -> {dict(candidate_modals)}", ) # ------------------------------------------------------------- # Enumeration preservation across the WHOLE candidate text. # A split/reordered candidate is allowed if every source list member # survives in token order somewhere in the complete candidate. # ------------------------------------------------------------- enumeration_groups = _enumeration_groups(analysis, original, source_doc) missing_by_group = [] for group_index, group in enumerate(enumeration_groups, start=1): missing = [member for member in group if not _member_present(member, revised)] if missing: missing_by_group.append({"group": group_index, "missing": missing}) checks["enumeration_groups"] = enumeration_groups checks["enumeration_missing"] = missing_by_group if missing_by_group: detail = " | ".join( f"group {entry['group']}: " + ", ".join(entry["missing"]) for entry in missing_by_group ) reject( "ENUMERATION_REJECTION", "Enumeration members missing: " + detail, ) # ------------------------------------------------------------- # Subject/object orientation # ------------------------------------------------------------- source_triples = _triples(source_doc) candidate_triples = _triples(candidate_doc) checks["source_triples"] = source_triples checks["candidate_triples"] = candidate_triples for subject, verb, obj in source_triples: if subject != obj and any( cs == obj and cq == subject and cv == verb for cs, cv, cq in candidate_triples ): reject( "SEMANTIC_REJECTION", f"Subject/object orientation reversed for '{verb}'.", ) if not source_triples and not candidate_triples: simple = ( r"(?iu)^\s*(?P[^\W\d_][\w-]*)\s+" r"(?P[^\W\d_][\w-]*)\s+(?P[^\W\d_][\w-]*)" ) source_match = re.match(simple, original) candidate_match = re.match(simple, revised) if ( source_match and candidate_match and source_match["s"].lower() == candidate_match["o"].lower() and source_match["o"].lower() == candidate_match["s"].lower() and source_match["v"].lower() == candidate_match["v"].lower() ): reject( "SEMANTIC_REJECTION", f"Subject/object orientation reversed for '{source_match['v']}'.", ) # ------------------------------------------------------------- # Discourse relations # ------------------------------------------------------------- source_relations = _relation_signature(original) candidate_relations = _relation_signature(revised) checks["relations"] = (sorted(source_relations), sorted(candidate_relations)) if "concession" in candidate_relations - source_relations: reject( "SEMANTIC_REJECTION", f"Candidate introduced concession: {sorted(source_relations)} -> {sorted(candidate_relations)}", ) if "cause" in source_relations and not ({"cause", "purpose"} & candidate_relations): reject( "SEMANTIC_REJECTION", f"Causal/result relation disappeared: {sorted(source_relations)} -> {sorted(candidate_relations)}", ) if "absence" in source_relations and "absence" not in candidate_relations: reject( "POLARITY_REJECTION", "The source's 'without' relation disappeared from the candidate.", ) if "dependency" in source_relations and "dependency" not in candidate_relations: reject( "SEMANTIC_REJECTION", "The source dependency/contingency relation disappeared.", ) # ------------------------------------------------------------- # Optional soft semantic signals # ------------------------------------------------------------- sts = stsim(original, revised) forward = nli_pair(original, revised) reverse = nli_pair(revised, original) checks.update(sts=sts, nli_forward=forward, nli_reverse=reverse) if sts is not None and sts < CFG.min_sts: checks["sts_warning"] = f"Low soft STS signal: {sts:.3f}" if forward and reverse: if max(forward["contradiction"], reverse["contradiction"]) > .55: reject("SEMANTIC_REJECTION", "Bidirectional NLI found contradiction.") elif min(forward["entailment"], reverse["entailment"]) < CFG.min_nli: checks["nli_warning"] = ( f"Weak entailment {forward['entailment']:.3f}/{reverse['entailment']:.3f}; " "manual review advised." ) # ------------------------------------------------------------- # Rewrite magnitude # ------------------------------------------------------------- surface = sr(original, revised) change = token_change(original, revised) checks.update(surface=surface, change=change, missing_terms=missing_terms) minimum_change = {"Minimal": .035, "Balanced": .065, "Structural": .09}[mode] if surface_key(original) == surface_key(revised): reject("INSUFFICIENT_REWRITE", "Candidate is identical to the source.") elif change < minimum_change: reject( "INSUFFICIENT_REWRITE", f"Token change {change:.3f} is below {mode} threshold {minimum_change:.3f}.", ) # ------------------------------------------------------------- # Optional local LanguageTool comparison # ------------------------------------------------------------- grammar_tool = optional("lt") if getattr(CFG, "lt", False) else None if grammar_tool: try: delta = len(grammar_tool.check(revised)) - len(grammar_tool.check(original)) checks["grammar_delta"] = delta if delta > 2: reject( "GRAMMAR_REJECTION", f"Candidate introduced {delta} additional LanguageTool findings.", ) except Exception as exc: checks["grammar_warning"] = str(exc) checks.update( failures=failures, rejection_reasons=reasons, accepted=not failures, ) return not failures, failures, checks # ===================================================================== # RANKER # ===================================================================== class Ranker: def rank(self, original, candidates, analysis, profile, user_terms, mode): good = [] bad = [] target = {"Minimal": .12, "Balanced": .24, "Structural": .38}[mode] for candidate in candidates: ok, failures, checks = Validator().validate( original, candidate, analysis, profile, user_terms, mode ) change = checks.get("change", token_change(original, candidate.text)) fit = max(0, 1 - abs(change - target) / max(target, .10)) sts = checks.get("sts") semantic = .78 if sts is None else max(0, min(1, sts)) nli_scores = [ item["entailment"] for item in [checks.get("nli_forward"), checks.get("nli_reverse")] if item ] if nli_scores: semantic = .55 * semantic + .45 * min(nli_scores) lane_fit = ( 1.0 if candidate.lane == mode else .82 if (mode, candidate.lane) in { ("Balanced", "Minimal"), ("Balanced", "Structural"), ("Structural", "Balanced"), } else .65 ) grammar = max(.6, 1 - .1 * max(0, checks.get("grammar_delta", 0))) style_penalties = _human_style_penalties(original, candidate.text) checks["human_style_penalties"] = style_penalties style_penalty = sum(float(item.get("weight", 0) or 0) for item in style_penalties) # Universal rules provide coverage for unseen vocabulary. A tiny # tie-break preference keeps the long-tested specialized rule when # both candidates are otherwise equivalent; it never changes hard # acceptance and does not prevent a universal candidate from winning. rule_kind = checks.get("candidate_rule_kind", "specialized") provenance_fit = .98 if rule_kind == "universal" else 1.0 candidate.score = ( 100 * ( .42 * semantic + .22 * fit + .15 * lane_fit + .12 * grammar + .07 * candidate.priority + .02 * provenance_fit - style_penalty ) if ok else -1e9 ) candidate.checks = checks (good if ok else bad).append(candidate) good.sort(key=lambda item: item.score, reverse=True) bad.sort(key=lambda item: ( len(item.checks.get("failures", [])), -item.checks.get("change", 0), )) return good, bad # ===================================================================== # ENGINE # ===================================================================== class Engine: def __init__(self): self.an = Analyzer() self.gen = Generator() self.sc = Scanner() self.rank = Ranker() def one(self, sentence, profile, user_terms, mode): _ensure_state() analysis = self.an.analyze(sentence, profile, user_terms) if analysis.families == ["PARSE_FAILURE"]: return SentenceResult( sentence, sentence, "PARSE_FAILURE", analysis.families, None, "Dependency parse unavailable and no surface-safe family matched.", [], {}, {"detected": analysis.families}, ) candidates, trace = self.gen.generate( analysis, profile, user_terms, mode ) if not candidates: code = ( "NO_PATTERN_COVERAGE" if not trace.get("routes") else "CANDIDATE_GENERATION_FAILURE" ) return SentenceResult( sentence, sentence, code, analysis.families, None, "No transformation produced a distinct candidate for the detected structures.", [], {}, trace, ) good, bad = self.rank.rank( sentence, candidates, analysis, profile, user_terms, mode ) trace["candidate_traces"] = [ { "text": item.text, "lane": item.lane, "family": item.family, "reason": item.reason, "accepted": item.checks.get("accepted"), "failures": item.checks.get("failures", []), "rejection_reasons": item.checks.get("rejection_reasons", []), "change": item.checks.get("change"), "sts": item.checks.get("sts"), "nli_forward": item.checks.get("nli_forward"), "nli_reverse": item.checks.get("nli_reverse"), "score": item.score, "rule_kind": item.checks.get("candidate_rule_kind"), "source_structure": item.checks.get("source_structure"), "candidate_structure": item.checks.get("candidate_structure"), "acronym_mapping_issues": item.checks.get("acronym_mapping_issues", []), "human_style_penalties": item.checks.get("human_style_penalties", []), "source_bounded_integrity_issues": item.checks.get("source_bounded_integrity_issues", []), } for item in good + bad ] if not good: counts = Counter( code for item in bad for code in item.checks.get("failures", []) ) precedence = [ "CONTENT_ADDITION_REJECTION", "REPETITION_REJECTION", "ACRONYM_MAPPING_REJECTION", "TERMINOLOGY_REJECTION", "POLARITY_REJECTION", "MODALITY_REJECTION", "QUANTITY_REJECTION", "ENUMERATION_REJECTION", "GRAMMAR_REJECTION", "SEMANTIC_REJECTION", "INSUFFICIENT_REWRITE", ] status = next((code for code in precedence if counts[code]), "RANKING_EMPTY") return SentenceResult( sentence, sentence, status, analysis.families, None, "Candidates were generated, but none passed all hard safeguards and rewrite gates.", [item.text for item in bad[:3]], bad[0].checks if bad else {}, trace | {"rejections": dict(counts)}, ) selected = good[0] trace["selected"] = { "text": selected.text, "lane": selected.lane, "family": selected.family, "score": selected.score, } return SentenceResult( sentence, selected.text, "SUCCESS", analysis.families, selected.family, selected.reason, [item.text for item in good[1:4]], selected.checks, trace, ) def process(self, text, workflow, mode, discipline, terms_text, progress=None): STATE.clear() STATE.update(start=time.time(), warnings=[]) profile, evidence = self.route(text, discipline) user_terms = { item.strip() for item in re.split(r"[,;\n]", terms_text or "") if item.strip() } findings, score, band, features = self.sc.scan(text) if workflow == "Scan Only": return DocumentResult( text, text, workflow, mode, profile.name, evidence, score, band, findings, [], STATE["warnings"], {"features": features}, ) paragraphs = [ item.strip() for item in re.split(r"\n\s*\n", text) if item.strip() ] total = sum(len(sent_split(item)) for item in paragraphs) done = 0 results = [] output_paragraphs = [] for block_start in range(0, len(paragraphs), CFG.batch): for paragraph in paragraphs[block_start:block_start + CFG.batch]: output_sentences = [] for sentence in sent_split(paragraph): targeted = ( workflow == "Edit / Paraphrase Only" or any(finding.sentence == len(results) for finding in findings) ) if targeted: item = self.one(sentence, profile, user_terms, mode) else: item = SentenceResult( sentence, sentence, "NOT_TARGETED", [], None, "No scanner finding targeted this sentence.", [], {}, {}, ) results.append(item) output_sentences.append(item.revised) done += 1 if progress: progress( done / max(total, 1), desc=f"Sentence {done}/{total}", ) output_paragraphs.append(" ".join(output_sentences)) return DocumentResult( text, "\n\n".join(output_paragraphs), workflow, mode, profile.name, evidence, score, band, findings, results, STATE["warnings"], { "sentences": total, "success": sum(item.status == "SUCCESS" for item in results), "unchanged": sum(item.original == item.revised for item in results), "seconds": time.time() - STATE["start"], }, ) def route(self, text, requested): if requested in PROFILES: return PROFILES[requested], ["manual"] lo = text.lower() scores = { name: sum( bool(re.search(r"(? 1) if _v5_duplicates: raise RuntimeError("Cell 7 v5.6 duplicate rule registration: " + ", ".join(_v5_duplicates)) if len(RULES) != EXPECTED_CELL7_RULE_COUNT: raise RuntimeError( f"Cell 7 v5.6 expected {EXPECTED_CELL7_RULE_COUNT} rules but found {len(RULES)}." ) # 11. Generic discipline-term mappings must remain stable. _term_source = ( "The process compared spectral response matrix (SRM), adaptive control " "protocol (ACP), and baseline measurements." ) _term_swapped_bad = Candidate( "The process compared spectral response matrix (ACP), adaptive control " "protocol (SRM), and baseline measurements.", "Balanced", "TEST", "generic acronym mapping self-test", .5, ) _term_analysis = _test_analyzer.analyze(_term_source, _test_profile, set()) _, _term_bad_fail, _term_bad_checks = Validator().validate( _term_source, _term_swapped_bad, _term_analysis, _test_profile, set(), "Balanced" ) if "ACRONYM_MAPPING_REJECTION" not in _term_bad_fail: raise RuntimeError( "Cell 7 v5.6 failed: swapped generic acronym mappings were not rejected: " + str(_term_bad_checks.get("acronym_mapping_issues")) ) # 12. Reordering complete atomic pairs is allowed. _term_reordered_good = Candidate( "The process compared adaptive control protocol (ACP), spectral response " "matrix (SRM), and baseline measurements.", "Balanced", "TEST", "generic acronym reorder self-test", .5, ) _, _term_good_fail, _term_good_checks = Validator().validate( _term_source, _term_reordered_good, _term_analysis, _test_profile, set(), "Balanced" ) if "ACRONYM_MAPPING_REJECTION" in _term_good_fail: raise RuntimeError( "Cell 7 v5.6 failed: intact reordered discipline-term pairs were rejected: " + str(_term_good_checks.get("acronym_mapping_issues")) ) # 13. Separating an acronym from its expansion is rejected. _term_separated_bad = Candidate( "The process compared SRM, adaptive control protocol (ACP), and baseline measurements.", "Balanced", "TEST", "generic acronym separation self-test", .5, ) _, _term_sep_fail, _ = Validator().validate( _term_source, _term_separated_bad, _term_analysis, _test_profile, set(), "Balanced" ) if "ACRONYM_MAPPING_REJECTION" not in _term_sep_fail: raise RuntimeError( "Cell 7 v5.6 failed: separated discipline-term pair was not rejected." ) # 14. Acronym discovery is case-sensitive and vocabulary-independent. if _acronym_inventory("The method used SRM, ACP, FGSM, and H2.") != Counter({"SRM":1,"ACP":1,"FGSM":1,"H2":1}): raise RuntimeError("Cell 7 v5.6 failed: generic acronym inventory is inaccurate.") if _acronym_inventory("The framework compared raw features."): raise RuntimeError("Cell 7 v5.6 failed: ordinary words were misclassified as acronyms.") if _semantic_tautology_issues("SRM is SRM.") != ["REDUNDANT_ACRONYM_COPULAR_TAUTOLOGY"]: raise RuntimeError("Cell 7 v5.6 failed: generic acronym tautology was not detected.") # 15. Awkward hypothesis attribution and new demonstratives are soft penalties. _style_probe = _human_style_penalties( "Android conditions did not remain within the thresholds; H2 was not supported.", "According to H2, Android conditions did not remain within the thresholds. " "This was not supported.", ) _style_codes = {item["code"] for item in _style_probe} if "AWKWARD_HYPOTHESIS_ATTRIBUTION" not in _style_codes: raise RuntimeError("Cell 7 v5.6 failed: awkward H2 attribution penalty missing.") if "NEW_DEMONSTRATIVE_ANTECEDENT" not in _style_codes: raise RuntimeError("Cell 7 v5.6 failed: demonstrative antecedent penalty missing.") # 16. Source-bounded non-generative integrity tests. _repeat_source = ( "The literature of history encompasses foundational narratives, " "transformative methodologies, and compelling modern accounts." ) _repeat_bad = ( "The literature of history encompasses foundational narratives. " "The literature of history encompasses transformative methodologies and compelling modern accounts." ) _repeat_issues = _source_bounded_integrity_issues(_repeat_source, _repeat_bad) _repeat_types = {item.get("type") for item in _repeat_issues} if "REPEATED_SENTENCE_OPENING_INTRODUCED" not in _repeat_types: raise RuntimeError( "Cell 7 v5.6 failed: repeated sentence opening evidence is missing: " + repr(_repeat_issues) ) _repeat_candidate = Candidate( _repeat_bad, "Structural", "TEST", "source-bounded repetition self-test", .5 ) _repeat_analysis = _test_analyzer.analyze(_repeat_source, _test_profile, set()) _, _repeat_failures, _repeat_checks = Validator().validate( _repeat_source, _repeat_candidate, _repeat_analysis, _test_profile, set(), "Structural", ) if "REPETITION_REJECTION" not in _repeat_failures: raise RuntimeError( "Cell 7 v5.6 failed: repeated output was not rejected by Validator: " + repr(_repeat_checks.get("source_bounded_integrity_issues")) ) _repeat_good = ( "Foundational narratives shape the field. " "Transformative methodologies explain later developments." ) if _source_bounded_integrity_issues(_repeat_source, _repeat_good): raise RuntimeError( "Cell 7 v5.6 failed: distinct sentence openings were falsely flagged: " + repr(_source_bounded_integrity_issues(_repeat_source, _repeat_good)) ) _exact_repeat_bad = "The model was evaluated. The model was evaluated." _exact_repeat_issues = _source_bounded_integrity_issues( "The model was evaluated.", _exact_repeat_bad ) if not any(item.get("type") == "REPEATED_SENTENCE_INTRODUCED" for item in _exact_repeat_issues): raise RuntimeError( "Cell 7 v5.6 failed: repeated sentence was not detected: " + repr(_exact_repeat_issues) ) _entity_addition_issues = _source_bounded_integrity_issues( "The study evaluated the classifier.", "The study evaluated the classifier at Microsoft Research.", ) if _named_entity_inventory("The study evaluated the classifier at Microsoft Research.") and not any( item.get("type") == "NAMED_ENTITY_INTRODUCED" for item in _entity_addition_issues ): raise RuntimeError( "Cell 7 v5.6 failed: introduced named entity was not detected: " + repr(_entity_addition_issues) ) # Legitimate deterministic restructuring must not be rejected merely because # function words or a relative pronoun were added. _source_bound_good = _source_bounded_integrity_issues( "Models predicting regional demand inform the allocation process.", "Models that predict regional demand inform the allocation process.", ) if _source_bound_good: raise RuntimeError( "Cell 7 v5.6 failed: safe grammatical scaffolding was rejected: " + repr(_source_bound_good) ) _pair_probe = _acronym_pairs( "The process compared spectral response matrix (SRM), adaptive control protocol (ACP), and baselines." ) _pair_probe_keys = { (pair["expansion_tokens"], pair["acronym"]) for pair in _pair_probe } _expected_pair_probe = { (("spectral", "response", "matrix"), "SRM"), (("adaptive", "control", "protocol"), "ACP"), } if _pair_probe_keys != _expected_pair_probe: raise RuntimeError( "Cell 7 v5.6 failed: generic atomic-pair extraction mismatch: " + repr(_pair_probe) ) print("=" * 78) print("CELL 7 UNIVERSAL VALIDATION / RANKING / ENGINE v5.6 LOADED") print("=" * 78) print("Validator: OK") print("Ranker: OK") print("Engine: OK") print("ENGINE instance: OK") print("Polarity-inventory validation: active") print("Dependency-aware Unicode enumeration validation: active") print("Appositive/list-boundary protection: active") print("Hyphen/dash-insensitive content comparison: active") print("Parser-artifact group filtering: active") print("Enumeration-member confidence gate: active") print("Noisy-superset group suppression: active") print("Source-bounded non-generative integrity: active") print("New named-entity addition rejection: active") print("Repeated sentence/opening/span rejection: active") print("Adaptive three-lemma sentence-opening detection: active") print("Validator-level repetition rejection self-test: PASSED") print("Content-expansion rejection: active") print("42-rule specialized + universal candidate judging: active") print("Vocabulary-independent structural signatures: active") print("Universal-rule ranking provenance: active") print("Generic atomic discipline-term extraction: active") print("No hard-coded acronym vocabulary: active") print("Acronym-definition mapping safeguard: active") print("Case-sensitive acronym discovery without hard-coded terms: active") print("Acronym separation/swap/duplication rejection: active") print("Acronym tautology rejection: active") print("Human-style hypothesis-attribution penalty: active") print("Demonstrative-antecedent ranking penalty: active") print("Unseen-vocabulary judge self-tests: PASSED") print("Cell 5A realization gate rechecked by judge: active") print("Optional STS/NLI signals: soft unless contradiction is detected") print("Runtime warning-state safety: active") print("Validator self-tests: PASSED") print("Registered transformation rules:", len(RULES)) print("Expected rule count:", EXPECTED_CELL7_RULE_COUNT) print("=" * 78)