diff --git a/aborist/qa/metacognition.py b/aborist/qa/metacognition.py new file mode 100644 index 0000000..7bb2b23 --- /dev/null +++ b/aborist/qa/metacognition.py @@ -0,0 +1,598 @@ +"""Meta-Cognition Preflight Guard — Ticket #000010 Phase 1. + +Runtime epistemic control layer that classifies a question's shape +BEFORE generation, so the model never answers from the surface form +alone when the question is ill-posed (false-premise, contradictory, +under-specified, broad-quantifier, time-sensitive, out-of-corpus, +reference-frame ambiguous). + +Pure and deterministic. No I/O, no model call, no retrieval call. +Reuses ``aborist.qa.quantifier.classify_question_quantifier`` for +the broad-quantifier rung; adds four new lightweight detectors: + + - temporal sensitivity (current/latest/today/CEO/etc.) + - contradiction (lexical) (unmarried+spouse, always+sometimes-not) + - false-premise (lite) (presupposition patterns) + - out-of-corpus (my-uploaded-X / my-file shapes) + +Reference-frame detection lives in ``aborist.qa.query._detect_frame`` +(ticket #000002) and is called from the surrounding runtime, not +from this module — keeps detection pure-on-question (no corpus +lookup needed here). + +The output is a ``QuestionState`` dataclass that becomes a CTI root +node. First pass surfaces it on the ``query()`` result dict only; +run-DAG node binding deferred to the same Phase 5 work tracked in +ticket #000009 (both nodes can land together). + +Hard rule (D1): No LLM in this hard path. Model-assisted preflight, +if added later, labels itself ``SOFT_PREFLIGHT_HINT`` and never +produces a ``PREFLIGHT_OK`` / ``PREFLIGHT_BLOCKED`` without +deterministic support. +""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import asdict, dataclass, field +from typing import Any, Literal + +PREFLIGHT_VERSION = "metacognition-v0.1" + + +# ---------------------------------------------------------------- types + +LogicalStatus = Literal[ + "well_formed", + "under_specified", + "false_premise_suspected", + "contradictory_question", + "out_of_corpus_risk", + "stale_risk", + "reference_frame_ambiguous", + "broad_quantifier_unbounded", +] + +PreflightResult = Literal[ + "PREFLIGHT_OK", + "PREFLIGHT_PARTIAL", + "PREFLIGHT_BLOCKED", +] + +TemporalSensitivity = Literal["high", "medium", "low"] + + +@dataclass(frozen=True) +class QuestionState: + """Runtime epistemic state for one question. + + All fields are deterministic from the question + per-call + `model_profile` / `corpus_profile` / `policy` inputs. No + randomness, no LLM. Hashable via `preflight_policy_hash` so + the run-DAG (Phase 5) can bind the decision into the audit + chain. + """ + + raw_question: str + question_hash: str + logical_statuses: tuple[LogicalStatus, ...] + question_shape: str + quantifier_intensity: str | None + quantifier_matched_token: str | None + scope_bound_hint: str + reference_frames: tuple[str, ...] + temporal_sensitivity: TemporalSensitivity + temporal_matched_tokens: tuple[str, ...] + contradiction_pairs: tuple[tuple[str, str], ...] + false_premise_hints: tuple[dict, ...] + corpus_requirement: str + known_boundaries: tuple[str, ...] + answer_constraints: dict + preflight_result: PreflightResult + preflight_policy_hash: str + classifier_version: str = PREFLIGHT_VERSION + + def to_dict(self) -> dict: + """Convert to JSON-serializable dict for run-DAG / bench.""" + out = asdict(self) + # asdict converts inner dataclasses but tuples-of-tuples + # come back as nested lists already — keep them as lists + # for JSON hygiene. + return out + + +# ---------------------------------------------------------------- temporal + +# Lexical patterns at start-of-question or as standalone tokens. +# `current`, `latest`, `today`, `now`, `as of`, `this year`. +_TEMPORAL_HIGH_PATTERNS = [ + re.compile(r"\bcurrent(?:ly)?\b", re.IGNORECASE), + re.compile(r"\blatest\b", re.IGNORECASE), + re.compile(r"\btoday\b", re.IGNORECASE), + re.compile(r"\bright now\b", re.IGNORECASE), + re.compile(r"\bas of\b", re.IGNORECASE), + re.compile(r"\bthis year\b", re.IGNORECASE), + re.compile(r"\bthis month\b", re.IGNORECASE), + re.compile(r"\bthis week\b", re.IGNORECASE), + re.compile(r"\brecently?\b", re.IGNORECASE), +] + +# Role-shape patterns: questions about who currently holds a role. +# Conservative — only positions with rapid turnover. +_TEMPORAL_ROLE_PATTERNS = [ + re.compile(r"\bCEO\b"), + re.compile(r"\bpresident of\b", re.IGNORECASE), + re.compile(r"\bprime minister\b", re.IGNORECASE), + re.compile(r"\bcurrent (?:champion|holder|price|stock)\b", re.IGNORECASE), +] + + +def detect_temporal_sensitivity( + question: str, +) -> tuple[TemporalSensitivity, tuple[str, ...]]: + """Return ``(sensitivity, matched_tokens)``. + + `high` = explicit temporal anchor (`current`, `latest`, etc.) + OR rapid-turnover role pattern. `medium` reserved for future + weekly/monthly cadence detection (not implemented in this + pass). `low` = no temporal markers detected (the default). + """ + matched: list[str] = [] + for pat in _TEMPORAL_HIGH_PATTERNS: + m = pat.search(question) + if m: + matched.append(m.group(0)) + for pat in _TEMPORAL_ROLE_PATTERNS: + m = pat.search(question) + if m: + matched.append(m.group(0)) + if matched: + return "high", tuple(matched) + return "low", () + + +# ---------------------------------------------------------------- contradiction + +# Conservative lexical-contradiction pairs. Only fires when BOTH +# tokens appear in the question. False positives are operator- +# hostile so we keep the list short and obvious. +_CONTRADICTION_PAIRS: tuple[tuple[str, str], ...] = ( + ("unmarried", "spouse"), + ("unmarried", "married"), + ("never", "always"), + ("alive", "dead"), + ("nonexistent", "existing"), + ("only", "also"), +) + + +def detect_contradiction( + question: str, +) -> tuple[tuple[str, str], ...]: + """Return tuple of (token_a, token_b) pairs whose BOTH members + appear in ``question`` (case-insensitive whole-word match). + + Returns empty tuple when no contradiction detected. The caller + decides whether to label-only or block — by default this + surfaces in the audit-line tail, NOT a hard block, since false + positives on contradiction would refuse legitimate questions. + """ + q_lower = question.lower() + found: list[tuple[str, str]] = [] + for a, b in _CONTRADICTION_PAIRS: + # Word-boundary match on each side independently. + a_re = re.compile(rf"\b{re.escape(a)}\b") + b_re = re.compile(rf"\b{re.escape(b)}\b") + if a_re.search(q_lower) and b_re.search(q_lower): + found.append((a, b)) + return tuple(found) + + +# ---------------------------------------------------------------- false premise + +# Presupposition patterns. Each pattern extracts an implied relation +# from the question shape. The verifier uses `false_premise_hints` +# as a `required_evidence` hint — the question is NOT blocked, but +# the audit-line tail surfaces "false premise suspected" so the +# operator knows the system didn't blindly accept the premise. + +# Subject and predicate character classes deliberately allow periods +# ("Mr.", "U.S."), apostrophes ("Homer's"), and hyphens +# ("by-law"). End-marker is `?` only (declarative variants of these +# question-shapes are not the target). +_FP_SUBJ = r"[\w\s\.\-']+?" +_FP_PRED = r"[\w\s\.\-']+?" + +_FALSE_PREMISE_PATTERNS = [ + # "when did X stop Y?" → presupposes X did Y + ( + re.compile( + rf"\bwhen did\s+(?P{_FP_SUBJ})\s+stop\s+(?P{_FP_PRED})\s*\?", + re.IGNORECASE, + ), + "stopped_doing", + "X did Y at some prior time", + ), + # "why did X cause Y?" → presupposes X caused Y + ( + re.compile( + rf"\bwhy did\s+(?P{_FP_SUBJ})\s+cause\s+(?P{_FP_PRED})\s*\?", + re.IGNORECASE, + ), + "caused", + "X caused Y", + ), + # "how did X become Y?" → presupposes X became Y + ( + re.compile( + rf"\bhow did\s+(?P{_FP_SUBJ})\s+become\s+(?P{_FP_PRED})\s*\?", + re.IGNORECASE, + ), + "became", + "X became Y", + ), + # "when did X become Y?" → presupposes X became Y + ( + re.compile( + rf"\bwhen did\s+(?P{_FP_SUBJ})\s+become\s+(?P{_FP_PRED})\s*\?", + re.IGNORECASE, + ), + "became", + "X became Y", + ), +] + + +def detect_false_premise(question: str) -> tuple[dict, ...]: + """Return tuple of presupposition dicts surfacing the implied + relation. Each dict carries: + + kind — pattern label (stopped_doing, caused, ...) + presupposition — natural-language statement of the + presupposition + subject — extracted subject token-span + predicate — extracted predicate token-span + + First-pass detection only. The verifier uses these as soft + hints; downstream the audit-line tail surfaces "false premise + suspected" so the operator can read the audit log and check + whether the cited evidence supports the presupposition. + + Returns empty tuple when no pattern fires. + """ + hints: list[dict] = [] + if not question: + return () + # Append a `?` if the question lacks one — the patterns + # require a sentence-ending marker for the predicate group. + test_q = question if question.rstrip().endswith(("?", ".")) else question + "?" + for pat, kind, presup_template in _FALSE_PREMISE_PATTERNS: + m = pat.search(test_q) + if m: + subject = m.group("subject").strip() + predicate = m.group("predicate").strip() + hints.append({ + "kind": kind, + "presupposition": presup_template.replace( + "X", subject + ).replace("Y", predicate), + "subject": subject, + "predicate": predicate, + }) + return tuple(hints) + + +# ---------------------------------------------------------------- out-of-corpus + +# Patterns that signal the operator is asking about a private / +# uploaded / non-corpus document. Conservative — defaults to +# "likely_in_corpus" for typical encyclopedic questions. +_OUT_OF_CORPUS_PATTERNS = [ + re.compile(r"\bmy (?:uploaded|unpublished|attached|private) [\w\s]+\b", re.IGNORECASE), + re.compile(r"\bthe file (?:i sent|i uploaded|i attached)\b", re.IGNORECASE), + re.compile(r"\bthe document (?:i sent|i uploaded|i attached)\b", re.IGNORECASE), + re.compile(r"\bin my (?:contract|email|notes|spreadsheet|inbox)\b", re.IGNORECASE), + re.compile(r"\bwhat does my [\w\s]+ say\b", re.IGNORECASE), +] + + +def detect_out_of_corpus(question: str) -> bool: + """Return True iff the question references a private / uploaded + document that the encyclopedic corpus cannot have.""" + for pat in _OUT_OF_CORPUS_PATTERNS: + if pat.search(question): + return True + return False + + +# ---------------------------------------------------------------- preflight + +def _question_hash(question: str) -> str: + """Stable SHA-256 of the raw question string. Lower-cased, + whitespace-normalized so trivial variants share a hash.""" + canon = re.sub(r"\s+", " ", question.strip().lower()) + return hashlib.sha256(canon.encode("utf-8")).hexdigest() + + +def _preflight_policy_hash(*, policy: dict | None, version: str) -> str: + """Hash of the policy fields that drive preflight behavior + + the classifier version. Bumping any of them invalidates prior + QuestionState records on lookup.""" + relevant = { + "metacognition_enabled": (policy or {}).get("metacognition_enabled", True), + "metacognition_temporal_check": (policy or {}).get( + "metacognition_temporal_check", True + ), + "metacognition_contradiction_check": (policy or {}).get( + "metacognition_contradiction_check", True + ), + "metacognition_false_premise_check": (policy or {}).get( + "metacognition_false_premise_check", True + ), + "metacognition_out_of_corpus_check": (policy or {}).get( + "metacognition_out_of_corpus_check", True + ), + "metacognition_block_on_contradiction": (policy or {}).get( + "metacognition_block_on_contradiction", False + ), + "version": version, + } + payload = "|".join(f"{k}={relevant[k]}" for k in sorted(relevant)) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _classify_question_shape( + quantifier_intensity: str | None, + temporal: TemporalSensitivity, + has_contradiction: bool, + has_false_premise: bool, + has_out_of_corpus: bool, +) -> str: + """Map the detector outputs onto a coarse shape mnemonic. + + Used by downstream policy + audit display so the operator can + eyeball the question type without parsing the full QuestionState. + """ + if has_out_of_corpus: + return "out_of_corpus" + if has_contradiction: + return "contradictory" + if has_false_premise: + return "presupposing" + if temporal == "high": + return "time_sensitive" + if quantifier_intensity in {"ALL", "COMPREHENSIVE", "OPEN_REQUEST"}: + return "broad_request" + if quantifier_intensity in {"SMALL_NUM_EXPLICIT", "COMPARATIVE_BOUND"}: + return "bounded_count" + if quantifier_intensity == "ABSENT": + return "negation" + if quantifier_intensity == "PROPORTIONAL": + return "proportional" + return "single_fact" + + +def preflight_question( + question: str, + *, + model_profile_id: str | None = None, + corpus_profile: dict | None = None, + reference_frames: tuple[str, ...] = (), + policy: dict | None = None, +) -> QuestionState: + """Classify ``question`` deterministically into a QuestionState. + + Pure function. Reuses the Phase 1 quantifier classifier + (#000008) plus four new lightweight detectors (temporal, + contradiction, false-premise-lite, out-of-corpus). + + `corpus_profile` is an optional dict carrying corpus boundary + metadata (e.g. ``{"corpus_latest_timestamp": "2003-05-16"}``); + when present, the temporal detector cross-checks against it. + First-pass implementation just records `corpus_requirement` + based on the temporal sensitivity — full cutoff arithmetic + deferred to a future amend. + + `reference_frames` is passed in by the caller because frame + detection requires retrieved sources (lives in + `aborist.qa.query._detect_frame`). Empty tuple is the default + for "no frame routing happened". + + `policy` overrides for the per-detector enables. Defaults are + permissive (all checks on) per ticket #000010 §7.3. + """ + from aborist.qa.quantifier import classify_question_quantifier + + policy = policy or {} + enabled = bool(policy.get("metacognition_enabled", True)) + temporal_on = bool(policy.get("metacognition_temporal_check", True)) + contradiction_on = bool(policy.get("metacognition_contradiction_check", True)) + false_premise_on = bool(policy.get("metacognition_false_premise_check", True)) + out_of_corpus_on = bool(policy.get("metacognition_out_of_corpus_check", True)) + block_on_contradiction = bool( + policy.get("metacognition_block_on_contradiction", False) + ) + + # Empty-question short-circuit. + if not question or not question.strip(): + return QuestionState( + raw_question=question or "", + question_hash=_question_hash(question or ""), + logical_statuses=(), + question_shape="empty", + quantifier_intensity=None, + quantifier_matched_token=None, + scope_bound_hint="unknown", + reference_frames=(), + temporal_sensitivity="low", + temporal_matched_tokens=(), + contradiction_pairs=(), + false_premise_hints=(), + corpus_requirement="not_applicable", + known_boundaries=("empty question",), + answer_constraints={}, + preflight_result="PREFLIGHT_BLOCKED", + preflight_policy_hash=_preflight_policy_hash( + policy=policy, version=PREFLIGHT_VERSION + ), + ) + + # Master kill — return a stub QuestionState with no detector + # output so the result schema stays consistent. Caller can + # distinguish "guard off" from "well-formed question" via the + # logical_statuses tuple being empty AND classifier_version. + if not enabled: + return QuestionState( + raw_question=question, + question_hash=_question_hash(question), + logical_statuses=(), + question_shape="metacognition_disabled", + quantifier_intensity=None, + quantifier_matched_token=None, + scope_bound_hint="unknown", + reference_frames=(), + temporal_sensitivity="low", + temporal_matched_tokens=(), + contradiction_pairs=(), + false_premise_hints=(), + corpus_requirement="not_evaluated", + known_boundaries=(), + answer_constraints={}, + preflight_result="PREFLIGHT_OK", + preflight_policy_hash=_preflight_policy_hash( + policy=policy, version=PREFLIGHT_VERSION + ), + ) + + # Reuse the #000008 quantifier classifier. + quant = classify_question_quantifier(question) + + # Run the four new detectors (each gateable). + if temporal_on: + temporal, temporal_matched = detect_temporal_sensitivity(question) + else: + temporal, temporal_matched = "low", () + + if contradiction_on: + contradictions = detect_contradiction(question) + else: + contradictions = () + + if false_premise_on: + false_premise = detect_false_premise(question) + else: + false_premise = () + + if out_of_corpus_on: + out_of_corpus = detect_out_of_corpus(question) + else: + out_of_corpus = False + + # Compose logical statuses. + statuses: list[LogicalStatus] = [] + if quant.get("is_broad") and quant.get("scope_bound_hint") == "unbounded": + statuses.append("broad_quantifier_unbounded") + elif quant.get("is_broad"): + statuses.append("under_specified") + if temporal == "high": + statuses.append("stale_risk") + if contradictions: + statuses.append("contradictory_question") + if false_premise: + statuses.append("false_premise_suspected") + if out_of_corpus: + statuses.append("out_of_corpus_risk") + if reference_frames and len(reference_frames) > 1: + statuses.append("reference_frame_ambiguous") + if not statuses: + statuses.append("well_formed") + + # Compose answer constraints. + answer_constraints: dict[str, Any] = {} + if "broad_quantifier_unbounded" in statuses or "under_specified" in statuses: + answer_constraints["bounded_or_reject"] = True + answer_constraints["max_claims_hint"] = quant.get("explicit_count") or 8 + if "stale_risk" in statuses: + answer_constraints["requires_current_source"] = True + if "false_premise_suspected" in statuses: + answer_constraints["require_premise_evidence"] = [ + h["presupposition"] for h in false_premise + ] + if "out_of_corpus_risk" in statuses: + answer_constraints["expected_corpus_status"] = "out_of_corpus" + + # Known boundaries — human-readable hints for the audit-line + # render layer + bench operator. + boundaries: list[str] = [] + if temporal_matched: + boundaries.append( + f"temporal markers: {', '.join(temporal_matched)}" + ) + if contradictions: + boundaries.append( + "lexical contradiction: " + + ", ".join(f"{a}/{b}" for a, b in contradictions) + ) + if false_premise: + boundaries.append( + "presuppositions: " + + "; ".join(h["presupposition"] for h in false_premise) + ) + if out_of_corpus: + boundaries.append("references private/uploaded document") + if quant.get("is_broad"): + boundaries.append( + f"broad quantifier ({quant.get('intensity')}, " + f"scope={quant.get('scope_bound_hint')})" + ) + + # Decide preflight result. + # PREFLIGHT_BLOCKED only when an explicit blocking condition is + # set in policy (default False for contradiction-block); otherwise + # PREFLIGHT_PARTIAL when any non-OK status fires; PREFLIGHT_OK + # when only "well_formed" is present. + if "well_formed" in statuses and len(statuses) == 1: + preflight_result: PreflightResult = "PREFLIGHT_OK" + elif "out_of_corpus_risk" in statuses: + preflight_result = "PREFLIGHT_BLOCKED" + elif block_on_contradiction and "contradictory_question" in statuses: + preflight_result = "PREFLIGHT_BLOCKED" + else: + preflight_result = "PREFLIGHT_PARTIAL" + + corpus_requirement = ( + "needs_current_source" + if temporal == "high" + else "out_of_corpus_likely" + if out_of_corpus + else "encyclopedic" + ) + + return QuestionState( + raw_question=question, + question_hash=_question_hash(question), + logical_statuses=tuple(statuses), + question_shape=_classify_question_shape( + quantifier_intensity=quant.get("intensity"), + temporal=temporal, + has_contradiction=bool(contradictions), + has_false_premise=bool(false_premise), + has_out_of_corpus=out_of_corpus, + ), + quantifier_intensity=quant.get("intensity"), + quantifier_matched_token=quant.get("matched_token"), + scope_bound_hint=quant.get("scope_bound_hint", "unknown"), + reference_frames=tuple(reference_frames), + temporal_sensitivity=temporal, + temporal_matched_tokens=tuple(temporal_matched), + contradiction_pairs=tuple(contradictions), + false_premise_hints=tuple(false_premise), + corpus_requirement=corpus_requirement, + known_boundaries=tuple(boundaries), + answer_constraints=answer_constraints, + preflight_result=preflight_result, + preflight_policy_hash=_preflight_policy_hash( + policy=policy, version=PREFLIGHT_VERSION + ), + ) diff --git a/docs/TICKETS.md b/docs/TICKETS.md index e423930..67b9e69 100644 --- a/docs/TICKETS.md +++ b/docs/TICKETS.md @@ -57,6 +57,7 @@ Newest first. Update on every open/close. | ID | Title | Status | Opened | Directive | |----------|------------------------------------------------|-----------------------|------------|-----------| +| #000010 | Meta-Cognition Preflight Guard (M0 / MCTL) | open · in progress (zero-shot 2026-05-03) | 2026-05-03 | D1, D3 | | #000009 | Quantifier preflight run-DAG node binding | open · awaiting go/no-go | 2026-05-03 | D3, D4 | | #000008 | Broad-quantifier preflight guard | closed · landed in `4f2b5a6`; Phase 5 DAG binding split into #000009 | 2026-05-02 | — | | #000007 | Query-layer hyphen folding | closed · 2026-05-02 | 2026-05-02 | — | @@ -69,4 +70,4 @@ Newest first. Update on every open/close. ## Next ID -`000010` +`000011` diff --git a/docs/tickets/ticket-000010-metacognition-preflight-guard.md b/docs/tickets/ticket-000010-metacognition-preflight-guard.md new file mode 100644 index 0000000..ff02bee --- /dev/null +++ b/docs/tickets/ticket-000010-metacognition-preflight-guard.md @@ -0,0 +1,324 @@ +# Ticket #000010 — Meta-Cognition Preflight Guard (M0 / MCTL) + +**Status:** open · in progress (zero-shot implementation 2026-05-03) +**Opened:** 2026-05-03 +**Scope:** Build a runtime epistemic control layer that classifies a +question's shape (false-premise, contradictory, under-specified, +broad-quantifier, time-sensitive, out-of-corpus, reference-frame +ambiguous) BEFORE generation. Forces Hermes to answer only through +the resulting `QuestionState` constraints rather than from surface +form alone. +**Audience:** fox + future blackops shifts. +**Source:** `~/Downloads/meta-cognition_for_hermes(1).txt` +(2026-05-03, Asia/Kuala_Lumpur). +**Hard constraint:** No `schema_version`, +`canonicalization_version`, or `chunking_version` bumps. Per #000008 +§1: pure additive policy folding into `governance_policy_hash`. No +LLM in hard preflight path — model-assisted preflight, if added, +labels itself `SOFT_PREFLIGHT_HINT` and never produces a +`PREFLIGHT_OK` / `PREFLIGHT_BLOCKED` without deterministic support +(seven-point program D1 — verifier never calls LLM). + +--- + +## 1. Premise + +A model cannot reliably infer the limits of its own knowledge from +inside generation. But a runtime can maintain explicit epistemic +state — corpus boundaries, source roots, retrieval coverage, model +profile, cutoff metadata, verifier policy, admissibility labels — +and force every answer through that state. + +The solution is not "ask Hermes to be more self-aware." The +solution is to build a meta-cognition preflight + epistemic +boundary guard around Hermes. + +This fits the existing stack: + +``` +Hermes proposes. +CTI structures. +Reverse-RAG binds claims to evidence. +Merkle-AGI-DAG commits the trace. +PROMETHEUS-Σ governs admissibility. +Meta-Cognition Preflight decides whether the question is +answerable, ill-posed, stale, unbounded, contradictory, or +outside corpus before generation. +``` + +## 2. What meta-cognition means here + +Not inner consciousness. A runtime control layer that answers +these questions BEFORE the model answers the user: + +- What kind of question is this? +- Is the question logically well-formed? +- Does it contain false or unproven premises? +- Is it bounded enough to answer? +- Does the current corpus contain the required evidence? +- Is the answer time-sensitive relative to model/source cutoff? +- Can the claim be grounded by retrieved evidence? +- Should the system answer, narrow scope, ask clarification, + reject, or label uncertainty? + +Call it **M0 — Meta-Cognition Preflight Guard** or in PROMETHEUS-Σ +language, **MCTL — Meta-Cognition Trigger Layer**. + +## 3. What cannot be solved completely + +You cannot fully prevent every illogical question or trap because: + +- Language is ambiguous; +- Unknown unknowns exist; +- Source corpora are incomplete; +- A false premise may be subtle; +- Retrieval may miss the correcting evidence; +- The model may know a true fact but the corpus may not support it; +- Some questions are undecidable, subjective, or under-specified; +- Time-sensitive facts can change after the corpus cutoff. + +So the target is not "never be trapped." The target is: + +> When trapped, fail visibly, locally, reproducibly, and non-reusably. + +In our labels: + +``` +ANSWERABLE +PARTIAL +UNSUPPORTED +OUT_OF_CORPUS +STALE_RISK +FALSE_PREMISE_SUSPECTED +CONTRADICTORY_QUESTION +UNDER_SPECIFIED +BROAD_QUANTIFIER_UNBOUNDED +REFERENCE_FRAME_AMBIGUOUS +NEEDS_CLARIFICATION +``` + +## 4. The core design + +Pre-generation layer: + +``` +User question +→ Meta-Cognition Preflight +→ CTI question-state object (QuestionState) +→ retrieval plan +→ evidence map +→ Hermes pointer-claim generation +→ deterministic verification +→ anchor/warrant checks +→ final admissibility label +→ Merkle-DAG commit +``` + +The preflight does not answer the question. It classifies the +question and sets constraints. + +## 5. New object: `QuestionState` + +```python +from dataclasses import dataclass +from typing import Literal + +LogicalStatus = Literal[ + "well_formed", + "under_specified", + "false_premise_suspected", + "contradictory_question", + "out_of_corpus_risk", + "stale_risk", + "reference_frame_ambiguous", + "broad_quantifier_unbounded", +] + +@dataclass(frozen=True) +class QuestionState: + raw_question: str + question_hash: str + logical_statuses: list[LogicalStatus] + question_shape: str + quantifier_intensity: str | None + reference_frames: list[str] + temporal_sensitivity: str + corpus_requirement: str + known_boundaries: list[str] + answer_constraints: dict + preflight_policy_hash: str +``` + +This object is a CTI root node and gets Merkle-committed (Phase 5 +follow-up; first pass surfaces it on the result dict only). + +## 6. Detector classes + +### 6.1 Quantifier (REUSE from #000008) + +Already shipped as `aborist/qa/quantifier.py`. Returns intensity + +`scope_bound_hint`. Meta-cognition wraps it; doesn't duplicate. + +### 6.2 Temporal sensitivity + +Lexical patterns: `current`, `latest`, `today`, `now`, `as of`, +`this year`, plus role-shape patterns: `CEO`, `president`, +`prime minister`, `current champion`, `current price`. Sets +`temporal_sensitivity ∈ {high, medium, low}`. When `high`, cross- +checks `model_profile.cutoff` and `corpus_latest_timestamp` against +`current_date` and emits `stale_risk` if cutoff is older. + +### 6.3 Contradiction detector (lexical pairs) + +``` +unmarried + spouse +always + sometimes not +never + did +only + also +before + after +married + unmarried +alive + dead +``` + +Conservative — only fires on clear lexical contradictions. False +positives are operator-hostile. + +### 6.4 False-premise detector (lite) + +Presupposition patterns: + +``` +when did X stop Y? → presupposes X did Y +why did X cause Y? → presupposes X caused Y +who is X's Y? → presupposes X has Y +which X is the real Y? → presupposes a real Y exists +how did X become Y? → presupposes X became Y +``` + +First pass: extract the implied relation as a `required_evidence` +hint that the verifier can use. Don't BLOCK the question — just +record `false_premise_suspected` so the audit-line can surface it. + +### 6.5 Reference-frame detector + +Reuse the existing frame-detection logic from ticket #000002 +(`aborist/qa/query.py:_detect_frame`). Already classifies +`reference` vs `literal` vs `ambiguous`. Meta-cognition wraps the +output as `reference_frames: list[str]`. + +### 6.6 Out-of-corpus detector + +Lexical patterns: `my unpublished`, `my uploaded`, `the file I +sent`, `what does it say about`. When detected on a corpus that +has no matching upload, sets `corpus_status: "out_of_corpus"`. +Conservative — defaults to `"likely_in_corpus"` for typical +encyclopedic questions. + +## 7. Implementation in Aborist + +### 7.1 New module + +`aborist/qa/metacognition.py` with: + +```python +def preflight_question( + question: str, + *, + model_profile_id: str | None = None, + corpus_profile: dict | None = None, + policy: dict | None = None, +) -> QuestionState: + ... +``` + +Pure and deterministic. Reuses the Phase 1 quantifier classifier +plus 4 new lightweight detectors. No I/O, no model call, no +retrieval call. + +### 7.2 Wire into query() / runner.ask() + +Same insertion pattern as the quantifier classifier: after policy +resolution, before retrieval. Result dict surfaces the +`QuestionState` so bench rows pick it up. + +### 7.3 Policy fields (additive) + +``` +metacognition_enabled True +metacognition_temporal_check True +metacognition_contradiction_check True +metacognition_false_premise_check True +metacognition_out_of_corpus_check True +metacognition_block_on_contradiction False (label-only by default) +``` + +All fold into `_VERIFIER_POLICY_FIELDS` so flipping any of them +invalidates prior cache records. + +### 7.4 Labels + +Composable with existing audit-line ladder. Examples: + +``` +POINTER-LINKED · PREFLIGHT_OK +ANCHOR-WARRANTED · PREFLIGHT_OK +EVIDENCE-WARRANTED · PREFLIGHT_OK +POINTER-LINKED · STALE_RISK +UNGROUNDED · FALSE_PREMISE_SUSPECTED +UNGROUNDED · CONTRADICTORY_QUESTION +UNGROUNDED · OUT_OF_CORPUS +``` + +## 8. Hard rule (D1) + +No LLM in preflight hard path. Hermes may propose +`SOFT_PREFLIGHT_HINT` (sidecar) but cannot create +`PREFLIGHT_OK` / `PREFLIGHT_BLOCKED` without deterministic support. +Preserves seven-point program D1 (verifier never calls LLM). + +## 9. Implementation order (per source doc §17) + +``` +1. metacognition.py with QuestionState. +2. Deterministic detectors: + quantifier (REUSE), temporal, reference-frame (REUSE), + contradiction, false-premise-lite, out-of-corpus. +3. Wire into query() / runner.ask(). +4. Add preflight labels to renderer. +5. Add policy fields to governance_policy_hash. +6. Add bench fields. +7. Add tests. +8. Run-DAG node binding (defer to Phase 5 ticket #000009 work — + same audit-replay gap; both nodes can land together). +9. Only later: model-assisted preflight as soft sidecar. +``` + +## 10. Test surface + +Per source doc §14: + +- False premise: `When did Mr. Burns become Homer's biological + father?` → `false_premise_suspected`. +- Contradiction: `Which unmarried spouse is Homer married to?` + → `contradictory_question`. +- Broad quantifier: `Winners of all major sports?` → + `broad_quantifier_unbounded`. +- Reference frame: `Has Oceania always been at war with East + Asia?` → reference frame includes orwell_1984. +- Time-sensitive: `Who is the current CEO of OpenAI?` → + `stale_risk` unless current source retrieved. +- Out-of-corpus: `What does my uploaded contract say about + clause 9?` → `out_of_corpus`. +- Model cutoff: model profile cutoff `2025-08`, current date + `2026-05-03`, question `latest winner of X` → + `requires_current_source`. + +## 11. Status + +Open · in progress 2026-05-03. Zero-shot implementation per fox +directive. Updates land here as phases ship; blockers escalate +to fox. + +## 12. Implementation log + +(populated as phases land) diff --git a/tests/test_metacognition.py b/tests/test_metacognition.py new file mode 100644 index 0000000..06d31e2 --- /dev/null +++ b/tests/test_metacognition.py @@ -0,0 +1,292 @@ +"""Meta-Cognition Preflight Guard — Ticket #000010 Phase 1 tests. + +Pin the seven test cases from the source doc §14 plus per-detector +unit tests. All checks are deterministic; no LLM, no I/O. + +Source: ``~/Downloads/meta-cognition_for_hermes(1).txt`` 2026-05-03. +""" + +from __future__ import annotations + +import pytest + +from aborist.qa.metacognition import ( + PREFLIGHT_VERSION, + QuestionState, + detect_contradiction, + detect_false_premise, + detect_out_of_corpus, + detect_temporal_sensitivity, + preflight_question, +) + + +# ---------------------------------------------------------------- temporal + +@pytest.mark.parametrize("q", [ + "Who is the current CEO of OpenAI?", + "What is the latest version of Python?", + "Who is the President of France today?", + "What is the current price of Bitcoin?", + "Who won the championship this year?", + "As of right now, who holds the world record?", +]) +def test_temporal_high_sensitivity(q): + sens, matched = detect_temporal_sensitivity(q) + assert sens == "high", f"{q} → {sens}" + assert len(matched) >= 1 + + +@pytest.mark.parametrize("q", [ + "what is the capital of france?", + "who painted the mona lisa?", + "what is the speed of light?", + "who wrote the play hamlet?", +]) +def test_temporal_low_sensitivity_for_factoids(q): + sens, _ = detect_temporal_sensitivity(q) + assert sens == "low" + + +# ---------------------------------------------------------------- contradiction + +def test_contradiction_unmarried_spouse(): + pairs = detect_contradiction("Which unmarried spouse is Homer married to?") + # `unmarried` + `spouse` AND `unmarried` + `married` both fire. + pair_set = set(pairs) + assert ("unmarried", "spouse") in pair_set + assert ("unmarried", "married") in pair_set + + +def test_contradiction_alive_dead(): + pairs = detect_contradiction("Which character is alive and dead simultaneously?") + assert ("alive", "dead") in set(pairs) + + +def test_contradiction_no_false_positives_on_singletons(): + """`unmarried` alone shouldn't fire — needs both halves of a pair.""" + pairs = detect_contradiction("Is George Washington still unmarried?") + assert pairs == () + + +def test_contradiction_clean_questions_have_no_pairs(): + pairs = detect_contradiction("who painted the mona lisa?") + assert pairs == () + + +# ---------------------------------------------------------------- false premise + +def test_false_premise_when_did_x_stop_y(): + hints = detect_false_premise("when did Mr. Burns stop being Homer's father?") + assert len(hints) >= 1 + h = hints[0] + assert h["kind"] == "stopped_doing" + assert "Mr. Burns" in h["subject"] + + +def test_false_premise_why_did_x_cause_y(): + hints = detect_false_premise("why did the moon cause the tides?") + assert len(hints) >= 1 + assert hints[0]["kind"] == "caused" + + +def test_false_premise_how_did_x_become_y(): + hints = detect_false_premise("how did Mr. Burns become Homer's biological father?") + assert len(hints) >= 1 + assert hints[0]["kind"] == "became" + assert "Mr. Burns" in hints[0]["subject"] + + +def test_false_premise_no_hint_on_neutral_question(): + hints = detect_false_premise("who is bilbo baggins?") + assert hints == () + + +# ---------------------------------------------------------------- out of corpus + +@pytest.mark.parametrize("q", [ + "What does my uploaded contract say about clause 9?", + "What is in my private notes from yesterday?", + "What does the file I sent you say about X?", + "In my email inbox, what did Alice write?", +]) +def test_out_of_corpus_detected(q): + assert detect_out_of_corpus(q) is True + + +@pytest.mark.parametrize("q", [ + "What is the capital of France?", + "Who wrote Hamlet?", + "What is the speed of light?", +]) +def test_out_of_corpus_not_detected_on_encyclopedic(q): + assert detect_out_of_corpus(q) is False + + +# ---------------------------------------------------------------- preflight (full integration) + +def test_preflight_factoid_is_well_formed(): + state = preflight_question("what is the capital of france?") + assert "well_formed" in state.logical_statuses + assert state.preflight_result == "PREFLIGHT_OK" + assert state.question_shape in ("single_fact", "negation") + + +def test_preflight_broad_quantifier_unbounded(): + """Source doc §5.3 + #000008 §10.1: `winners of all major sports?` + classifies as broad-quantifier-unbounded.""" + state = preflight_question("Winners of all major sports?") + assert "broad_quantifier_unbounded" in state.logical_statuses + assert state.preflight_result == "PREFLIGHT_PARTIAL" + assert state.quantifier_intensity == "ALL" + assert state.scope_bound_hint == "unbounded" + + +def test_preflight_broad_quantifier_bounded_does_not_fire_unbounded(): + """`name all members of the beatles` is a bounded universal. + Should classify as `under_specified` (broad but bounded), NOT + `broad_quantifier_unbounded`.""" + state = preflight_question("name all members of the beatles") + assert "broad_quantifier_unbounded" not in state.logical_statuses + assert state.scope_bound_hint == "bounded" + + +def test_preflight_contradictory_question(): + """Source doc §5.2: `Which unmarried spouse is Homer married to?` + classifies as contradictory.""" + state = preflight_question("Which unmarried spouse is Homer married to?") + assert "contradictory_question" in state.logical_statuses + assert state.contradiction_pairs # at least one pair + + +def test_preflight_false_premise_suspected(): + """Source doc §5.1: `When did Mr. Burns become Homer's biological + father?` classifies as false_premise_suspected.""" + state = preflight_question( + "When did Mr. Burns become Homer's biological father?" + ) + assert "false_premise_suspected" in state.logical_statuses + assert state.false_premise_hints + + +def test_preflight_time_sensitive_stale_risk(): + """Source doc §5.4: `Who is the current CEO of X?` classifies + as stale_risk.""" + state = preflight_question("Who is the current CEO of OpenAI?") + assert "stale_risk" in state.logical_statuses + assert state.temporal_sensitivity == "high" + assert state.answer_constraints.get("requires_current_source") is True + + +def test_preflight_out_of_corpus_blocked(): + """Source doc §5.5: out-of-corpus references should BLOCK.""" + state = preflight_question("What does my uploaded contract say?") + assert "out_of_corpus_risk" in state.logical_statuses + assert state.preflight_result == "PREFLIGHT_BLOCKED" + + +def test_preflight_reference_frame_ambiguous(): + """Source doc §5.6: when caller passes multiple reference frames, + classifier marks reference_frame_ambiguous.""" + state = preflight_question( + "Has Oceania always been at war with East Asia?", + reference_frames=("literal_geography", "orwell_1984"), + ) + assert "reference_frame_ambiguous" in state.logical_statuses + + +# ---------------------------------------------------------------- gating + +def test_master_kill_disables_preflight(): + """policy={'metacognition_enabled': False} short-circuits with + a stub QuestionState. Logical statuses tuple is empty; + preflight_result is PREFLIGHT_OK so downstream isn't blocked.""" + state = preflight_question( + "Winners of all major sports?", + policy={"metacognition_enabled": False}, + ) + assert state.logical_statuses == () + assert state.preflight_result == "PREFLIGHT_OK" + assert state.question_shape == "metacognition_disabled" + + +def test_per_detector_disable(): + """Each detector can be turned off independently. Verify the + contradiction check obeys its switch.""" + state = preflight_question( + "Which unmarried spouse is Homer married to?", + policy={"metacognition_contradiction_check": False}, + ) + # Contradiction detector skipped → no contradictory_question status. + assert "contradictory_question" not in state.logical_statuses + assert state.contradiction_pairs == () + + +def test_block_on_contradiction_opt_in(): + """By default contradictory questions surface as PARTIAL (label + only). `metacognition_block_on_contradiction=True` flips to + BLOCKED.""" + state_default = preflight_question( + "Which unmarried spouse is Homer married to?", + ) + state_strict = preflight_question( + "Which unmarried spouse is Homer married to?", + policy={"metacognition_block_on_contradiction": True}, + ) + assert state_default.preflight_result == "PREFLIGHT_PARTIAL" + assert state_strict.preflight_result == "PREFLIGHT_BLOCKED" + + +# ---------------------------------------------------------------- determinism + +def test_question_hash_stable(): + """Same question (modulo case + whitespace) → same hash.""" + a = preflight_question("Who painted the Mona Lisa?") + b = preflight_question(" who painted the mona lisa? ") + assert a.question_hash == b.question_hash + + +def test_preflight_policy_hash_changes_with_policy_flip(): + """Flipping a metacognition policy field bumps the policy hash — + enables governance binding via _VERIFIER_POLICY_FIELDS (Phase 3).""" + a = preflight_question( + "Winners of all major sports?", + policy={"metacognition_temporal_check": True}, + ) + b = preflight_question( + "Winners of all major sports?", + policy={"metacognition_temporal_check": False}, + ) + assert a.preflight_policy_hash != b.preflight_policy_hash + + +# ---------------------------------------------------------------- versioning + +def test_preflight_version_pinned(): + state = preflight_question("any question") + assert state.classifier_version == PREFLIGHT_VERSION + assert PREFLIGHT_VERSION == "metacognition-v0.1" + + +# ---------------------------------------------------------------- empty / edge + +def test_empty_question_blocked(): + state = preflight_question("") + assert state.preflight_result == "PREFLIGHT_BLOCKED" + assert state.question_shape == "empty" + + +def test_whitespace_only_question_blocked(): + state = preflight_question(" \n\t ") + assert state.preflight_result == "PREFLIGHT_BLOCKED" + + +def test_question_state_is_serializable(): + """to_dict() must produce JSON-friendly output for run-DAG / + bench JSONL persistence.""" + import json + state = preflight_question("Winners of all major sports?") + payload = state.to_dict() + # Round-trip through json.dumps; failure here means a non- + # serializable type leaked in. + json.dumps(payload, ensure_ascii=False)