lighthouse_v1_7.py

The PUBLIC LIGHTHOUSE CORE v1.7 runtime has been released, adding two new sublayers to Layer 3 (State Validation): the Orientation State Register (OSR) for spatial frame and basis tracking, and Narrative State Validation (NSV) for interpretation drift tracking. The update introduces new reason codes for detecting orientation frame drift, basis mismatches, and narrative reinterpretation issues, while leaving all other modules and the aggregator unchanged from v1.6. The system can be run with six demo modes including clean, overclaim, spatial_ambiguous, spatial_locked, narrative_drift, and narrative_retro.

| /usr/bin/env python3 | | | """ | | | PUBLIC LIGHTHOUSE CORE v1.7 | | | AI Output Governance and Validation Runtime | | | Adds two new sublayers to Layer 3 State Validation : | | | 3A State Validation drift scoring — unchanged from v1.6 | | | 3B Orientation State Register OSR — spatial frame / basis tracking | | | 3C Narrative State Validation NSV — interpretation drift tracking | | | All other modules and the aggregator are unchanged from v1.6. | | | Run: | | | python PUBLIC LIGHTHOUSE CORE v1 7.py --demo clean --pretty | | | python PUBLIC LIGHTHOUSE CORE v1 7.py --demo overclaim --pretty | | | python PUBLIC LIGHTHOUSE CORE v1 7.py --demo spatial ambiguous --pretty | | | python PUBLIC LIGHTHOUSE CORE v1 7.py --demo spatial locked --pretty | | | python PUBLIC LIGHTHOUSE CORE v1 7.py --demo narrative drift --pretty | | | python PUBLIC LIGHTHOUSE CORE v1 7.py --demo narrative retro --pretty | | | """ | | | from future import annotations | | | import argparse | | | from dataclasses import dataclass, field, asdict | | | from datetime import datetime, timezone | | | from enum import Enum | | | import json | | | from typing import Any, Dict, List, Optional | | | import uuid | | | SYSTEM VERSION = "PUBLIC LIGHTHOUSE CORE 1.7.0" | | | REASON REGISTRY VERSION = "1.1.0" | | | ── Decision types ───────────────────────────────────────────────────────── | | | class FinalDecision str, Enum : | | | AUTHORIZE = "AUTHORIZE" | | | CONSTRAIN = "CONSTRAIN" | | | REVISE = "REVISE" | | | HOLD = "HOLD" | | | CLARIFY = "CLARIFY" | | | HALT = "HALT" | | | BLOCK = "BLOCK" | | | FINAL PRIORITY = { | | | FinalDecision.BLOCK: 7, | | | FinalDecision.HALT: 6, | | | FinalDecision.CLARIFY: 5, | | | FinalDecision.HOLD: 4, | | | FinalDecision.REVISE: 3, | | | FinalDecision.CONSTRAIN: 2, | | | FinalDecision.AUTHORIZE: 1, | | | } | | | ── Reason codes ─────────────────────────────────────────────────────────── | | | class ReasonCode: | | | Preflight | | | PREFLIGHT PURPOSE MISSING = "PREFLIGHT PURPOSE MISSING" | | | PREFLIGHT BIASED FRAME = "PREFLIGHT BIASED FRAME" | | | PREFLIGHT FORCED CONCLUSION = "PREFLIGHT FORCED CONCLUSION" | | | Execution Control | | | EXECUTION AGENT UNVERIFIABLE = "EXECUTION AGENT UNVERIFIABLE" | | | EXECUTION SCOPE VIOLATION = "EXECUTION SCOPE VIOLATION" | | | EXECUTION PERMISSION MISSING = "EXECUTION PERMISSION MISSING" | | | EXECUTION STALE HALT = "EXECUTION STALE HALT" | | | Input Validation | | | INPUT FRAME NOT DECLARED = "INPUT FRAME NOT DECLARED" | | | INPUT KNOWN UNKNOWNS MISSING = "INPUT KNOWN UNKNOWNS MISSING" | | | INPUT IMPOSSIBLE REQUEST = "INPUT IMPOSSIBLE REQUEST" | | | Structural Validation | | | STRUCTURE CHAIN NOT REPLAYABLE = "STRUCTURE CHAIN NOT REPLAYABLE" | | | STRUCTURE TERMINAL STATE MISMATCH = "STRUCTURE TERMINAL STATE MISMATCH" | | | STRUCTURE STEP INVALID = "STRUCTURE STEP INVALID" | | | State Validation 3A | | | STATE INVARIANT VIOLATION = "STATE INVARIANT VIOLATION" | | | STATE BOUNDARY CROSSING = "STATE BOUNDARY CROSSING" | | | STATE SHADOW UNAVAILABLE = "STATE SHADOW UNAVAILABLE" | | | STATE IDENTITY LOST = "STATE IDENTITY LOST" | | | Orientation State Register 3B | | | ORIENTATION STATE MISSING = "ORIENTATION STATE MISSING" | | | ORIENTATION BASIS UNDECLARED = "ORIENTATION BASIS UNDECLARED" | | | ORIENTATION BASIS MISMATCH = "ORIENTATION BASIS MISMATCH" | | | ORIENTATION ATTACHMENT UNDECLARED = "ORIENTATION ATTACHMENT UNDECLARED" | | | ORIENTATION TRANSFORM UNSUPPORTED = "ORIENTATION TRANSFORM UNSUPPORTED" | | | ORIENTATION FRAME DRIFT = "ORIENTATION FRAME DRIFT" | | | ORIENTATION MOVEMENT BEFORE ROTATION = "ORIENTATION MOVEMENT APPLIED BEFORE ROTATION" | | | Narrative State Validation 3C | | | NARRATIVE OBSERVATION INTERPRETATION MERGED = "NARRATIVE OBSERVATION INTERPRETATION MERGE" | | | NARRATIVE CONFIDENCE ESCALATION UNDECLARED = "NARRATIVE CONFIDENCE ESCALATION UNDECLARED" | | | NARRATIVE REINTERPRETATION UNATTRIBUTED = "NARRATIVE REINTERPRETATION UNATTRIBUTED" | | | NARRATIVE CONFIDENCE EVIDENCE DECOUPLED = "NARRATIVE CONFIDENCE EVIDENCE DECOUPLED" | | | RETROACTIVE NARRATIVE COLLAPSE = "RETROACTIVE NARRATIVE COLLAPSE" | | | Risk Analysis | | | RISK IRREVERSIBLE LOW CONFIDENCE = "RISK IRREVERSIBLE LOW CONFIDENCE" | | | RISK CASCADING FAILURE SURFACE = "RISK CASCADING FAILURE SURFACE" | | | RISK FRAGILE ASSUMPTION = "RISK FRAGILE ASSUMPTION" | | | RISK UNRESOLVED CRITICAL = "RISK UNRESOLVED CRITICAL" | | | Communication Validation | | | COMMUNICATION TONE OVERCLAIM = "COMMUNICATION TONE OVERCLAIM" | | | COMMUNICATION COMPRESSION LOSS = "COMMUNICATION COMPRESSION LOSS" | | | COMMUNICATION IDENTITY BINDING = "COMMUNICATION IDENTITY BINDING" | | | COMMUNICATION FALSE CERTAINTY = "COMMUNICATION FALSE CERTAINTY" | | | Evidence Validation | | | EVIDENCE CAUSALITY OVERCLAIM = "EVIDENCE CAUSALITY OVERCLAIM" | | | EVIDENCE CONFIDENCE AS TRUTH = "EVIDENCE CONFIDENCE AS TRUTH" | | | EVIDENCE DASHBOARD AUTHORITY = "EVIDENCE DASHBOARD AUTHORITY" | | | EVIDENCE NO BASELINE = "EVIDENCE NO BASELINE" | | | System | | | SCOPE VIOLATION MODULE OVERREACH = "SCOPE VIOLATION MODULE OVERREACH" | | | REVISE LOOP UNRESOLVED = "REVISE LOOP UNRESOLVED" | | | CONSTRAINT INVALID = "CONSTRAINT INVALID" | | | ── Data contracts ───────────────────────────────────────────────────────── | | | @dataclass | | | class Reason: | | | code: str | | | module: str | | | severity: str | | | signal source: str = "" | | | description: str = "" | | | @dataclass | | | class Constraint: | | | source module: str | | | scope: str | | | bound: str | | | reason code: str | | | severity: str = "CONSTRAIN" | | | @dataclass | | | class ScopeViolation: | | | module: str | | | violation: str | | | correct module: str | | | reason code: str = ReasonCode.SCOPE VIOLATION MODULE OVERREACH | | | @dataclass | | | class ModuleResult: | | | module: str | | | decision: str | | | reasons: List Reason = field default factory=list | | | constraints: List Constraint = field default factory=list | | | scope violations: List ScopeViolation = field default factory=list | | | data: Dict str, Any = field default factory=dict | | | @dataclass | | | class RunContext: | | | pass number: int = 1 | | | max passes: int = 2 | | | revise loop active: bool = False | | | prior stable state id: Optional str = None | | | prediction model: str = "none" | | | halt resumed: bool = False | | | @dataclass | | | class TaskInput: | | | prompt: str | | | proposed output: str = "" | | | agent id: Optional str = "local user" | | | permission envelope: Optional Dict str, Any = field default factory=lambda: { | | | "declared capabilities": "analyze", "validate" , | | | "declared authority boundary": "local", | | | "allowed actions": "analyze", "validate" , | | | "resource scope": "local", | | | } | | | action: str = "analyze" | | | intent: Optional str = None | | | frame type: Optional str = None | | | known unknowns: List str = field default factory=list | | | evidence: Dict str, Any = field default factory=dict | | | transform chain: List Dict str, Any = field default factory=list | | | state: Dict str, Any = field default factory=dict | | | OSR fields Layer 3B | | | orientation: Optional Dict str, Any = None | | | NSV fields Layer 3C | | | narrative: Optional Dict str, Any = None | | | risk: Dict str, Any = field default factory=dict | | | metadata: Dict str, Any = field default factory=dict | | | ── Helpers ──────────────────────────────────────────────────────────────── | | | def utc now - str: | | | return datetime.now timezone.utc .isoformat | | | def reason code, module, severity, signal source="", description="" - Reason: | | | return Reason code=code, module=module, severity=severity, | | | signal source=signal source, description=description | | | def constraint module, scope, bound, code, severity="CONSTRAIN" - Constraint: | | | return Constraint source module=module, scope=scope, bound=bound, | | | reason code=code, severity=severity | | | def contains any text: str, terms: List str - bool: | | | lower = text.lower | | | return any t.lower in lower for t in terms | | | def confidence word text: str - bool: | | | return contains any text, "clearly", "definitely", "proves", "confirmed", | | | "guaranteed", "always", "never", "certain" | | | def causal word text: str - bool: | | | return contains any text, "because", "caused", "causes", "proves", | | | "driven by", "confirms" | | | def dashboard authority word text: str - bool: | | | return contains any text, "green", "score", "rank", "safe to automate", | | | "trustworthy", "authority" | | | ── Layer -1: Preflight ──────────────────────────────────────────────────── | | | class PreflightModule: | | | name = "preflight" | | | def run self, task: TaskInput - ModuleResult: | | | text = task.prompt or "" | | | if not task.intent: | | | if any q in text.lower for q in "evaluate", "analyze", "calculate", "what is", "should", "question" : | | | task.intent = "analysis" | | | else: | | | return ModuleResult self.name, "CLARIFY", | | | reason ReasonCode.PREFLIGHT PURPOSE MISSING, | | | self.name, "CLARIFY", "intent missing" | | | if contains any text, "convince me", "prove that", "show why this is true", "make them believe" : | | | return ModuleResult self.name, "CONSTRAIN", | | | reason ReasonCode.PREFLIGHT BIASED FRAME, | | | self.name, "CONSTRAIN", "steered or persuasive frame" , | | | constraint self.name, "output type", "analysis only no persuasion", | | | ReasonCode.PREFLIGHT BIASED FRAME | | | if contains any text, "ignore evidence", "force conclusion", "must conclude" : | | | return ModuleResult self.name, "BLOCK", | | | reason ReasonCode.PREFLIGHT FORCED CONCLUSION, | | | self.name, "BLOCK", "forced conclusion" | | | return ModuleResult self.name, "PROCEED" | | | ── Layer 0: Execution Control ───────────────────────────────────────────── | | | class ExecutionControlModule: | | | name = "execution control" | | | def run self, task: TaskInput - ModuleResult: | | | if not task.agent id: | | | return ModuleResult self.name, "HALT", | | | reason ReasonCode.EXECUTION AGENT UNVERIFIABLE, | | | self.name, "HALT", "missing agent id" | | | if not task.permission envelope: | | | return ModuleResult self.name, "DEFER", | | | reason ReasonCode.EXECUTION PERMISSION MISSING, | | | self.name, "DEFER", "missing permission envelope" | | | allowed = set task.permission envelope.get "allowed actions", | | | scope = task.permission envelope.get "resource scope", "" | | | boundary = task.permission envelope.get "declared authority boundary", "" | | | if task.action not in allowed: | | | return ModuleResult self.name, "BLOCK", | | | reason ReasonCode.EXECUTION SCOPE VIOLATION, | | | self.name, "BLOCK", "action not allowed" | | | if scope not in "local", "system", "external" : | | | return ModuleResult self.name, "BLOCK", | | | reason ReasonCode.EXECUTION SCOPE VIOLATION, | | | self.name, "BLOCK", "invalid resource scope" | | | if boundary == "local" and scope == "external": | | | return ModuleResult self.name, "BLOCK", | | | reason ReasonCode.EXECUTION SCOPE VIOLATION, | | | self.name, "BLOCK", "outside authority boundary" | | | return ModuleResult self.name, "READY" | | | ── Layer 1: Input Validation ────────────────────────────────────────────── | | | class InputValidationModule: | | | name = "input validation" | | | def run self, task: TaskInput - ModuleResult: | | | text = task.prompt or "" | | | if not text.strip : | | | return ModuleResult self.name, "CLARIFY", | | | reason ReasonCode.INPUT FRAME NOT DECLARED, | | | self.name, "CLARIFY", "empty prompt" | | | if contains any text, "square circle", "rotate 90 in two directions at once" : | | | return ModuleResult self.name, "BLOCK", | | | reason ReasonCode.INPUT IMPOSSIBLE REQUEST, | | | self.name, "BLOCK", "impossible request" | | | if not task.frame type: | | | task.frame type = "factual" if "what is" in text.lower else "analytical" | | | if task.frame type in "speculative", "operational" and not task.known unknowns: | | | return ModuleResult self.name, "CLARIFY", | | | reason ReasonCode.INPUT KNOWN UNKNOWNS MISSING, | | | self.name, "CLARIFY", "known unknowns missing" | | | return ModuleResult self.name, "PROCEED" | | | ── Layer 2: Structural Validation ───────────────────────────────────────── | | | class StructuralValidationModule: | | | name = "structure" | | | def run self, task: TaskInput - ModuleResult: | | | chain = task.transform chain | | | if not chain: | | | return ModuleResult self.name, "PASS", | | | data={"replayable": None, "note": "no explicit transform chain supplied"} | | | previous after = None | | | for i, step in enumerate chain : | | | missing = k for k in "operation", "state before", "state after" if k not in step | | | if missing: | | | return ModuleResult self.name, "FAIL", | | | reason ReasonCode.STRUCTURE CHAIN NOT REPLAYABLE, | | | self.name, "FAIL", f"missing step {i}" , | | | data={"missing step index": i, "missing fields": missing} | | | if previous after is not None and step "state before" = previous after: | | | return ModuleResult self.name, "FAIL", | | | reason ReasonCode.STRUCTURE STEP INVALID, | | | self.name, "FAIL", f"step {i} before mismatch" | | | previous after = step "state after" | | | claimed = task.metadata.get "claimed final state" | | | if claimed is not None and previous after is not None and claimed = previous after: | | | return ModuleResult self.name, "BLOCK", | | | reason ReasonCode.STRUCTURE TERMINAL STATE MISMATCH, | | | self.name, "BLOCK", "claimed final state mismatch" | | | return ModuleResult self.name, "PASS", data={"replayable": True} | | | ── Layer 3A: State Validation ───────────────────────────────────────────── | | | class StateValidationModule: | | | name = "state" | | | def run self, task: TaskInput, context: RunContext - ModuleResult: | | | state = task.state or {} | | | iv = int state.get "invariant violations", 0 | | | bc = int state.get "boundary crossings", 0 | | | ac = int state.get "assumption changes", 0 | | | ic = int state.get "inconsistencies", 0 | | | pd = bool state.get "prediction divergence", False and context.prediction model = "none" | | | shadow avail = bool state.get "shadow state available", False | | | needs shadow = bool state.get "requires shadow state", False | | | inv c = 0.30 1 if iv 0 else 0 | | | bnd c = 0.25 min bc / 2, 1.0 | | | asm c = 0.15 min ac / 3, 1.0 | | | inc c = 0.20 min ic / 2, 1.0 | | | prd c = 0.10 1 if pd else 0 | | | shd p = 0.05 if needs shadow and not shadow avail else 0.0 | | | drift = min inv c + bnd c + asm c + inc c + prd c + shd p, 1.0 | | | reasons = | | | if iv: | | | reasons.append reason ReasonCode.STATE INVARIANT VIOLATION, | | | self.name, "CONSTRAIN", "invariant violations" | | | if bc: | | | reasons.append reason ReasonCode.STATE BOUNDARY CROSSING, | | | self.name, "CONSTRAIN", "boundary crossings" | | | if needs shadow and not shadow avail: | | | reasons.append reason ReasonCode.STATE SHADOW UNAVAILABLE, | | | self.name, "INFO", "shadow state unavailable" | | | if state.get "identity lost", False : | | | reasons.append reason ReasonCode.STATE IDENTITY LOST, | | | self.name, "BLOCK", "identity lost" | | | return ModuleResult self.name, "BLOCK", reasons, | | | data=self. data drift, inv c, bnd c, asm c, inc c, prd c, "UNSTABLE" | | | if drift = 0.70: | | | cl = "UNSTABLE" | | | elif drift = 0.30: | | | cl = "CONSTRAINED" | | | else: | | | cl = "STABLE" | | | return ModuleResult self.name, cl, reasons, | | | data=self. data drift, inv c, bnd c, asm c, inc c, prd c, cl | | | @staticmethod | | | def data score, inv, bnd, asm, inc, prd, cl - Dict str, Any : | | | return { | | | "drift score": round score, 4 , | | | "drift components": { | | | "invariant": round inv, 4 , | | | "boundary": round bnd, 4 , | | | "assumption": round asm, 4 , | | | "inconsistency": round inc, 4 , | | | "prediction": round prd, 4 , | | | }, | | | "stability classification": cl, | | | } | | | ── Layer 3B: Orientation State Register ─────────────────────────────────── | | | Standard basis lookup table top-down / right-hand convention | | | ORIENTATION TABLE: Dict str, Dict str, Any = { | | | "IDENTITY": { | | | "local x": "global +X", | | | "local y": "global +Y", | | | "local z": "global +Z", | | | "basis vector": {"x": 1, 0, 0 , "y": 0, 1, 0 , "z": 0, 0, 1 }, | | | }, | | | "Y CW 90 TOP DOWN": { | | | "local x": "global -Z", | | | "local y": "global +Y", | | | "local z": "global +X", | | | "basis vector": {"x": 0, 0, -1 , "y": 0, 1, 0 , "z": 1, 0, 0 }, | | | }, | | | "Y CCW 90 TOP DOWN": { | | | "local x": "global +Z", | | | "local y": "global +Y", | | | "local z": "global -X", | | | "basis vector": {"x": 0, 0, 1 , "y": 0, 1, 0 , "z": -1, 0, 0 }, | | | }, | | | "Y 180": { | | | "local x": "global -X", | | | "local y": "global +Y", | | | "local z": "global -Z", | | | "basis vector": {"x": -1, 0, 0 , "y": 0, 1, 0 , "z": 0, 0, -1 }, | | | }, | | | "X CW 90 RIGHT SIDE": { | | | "local x": "global +X", | | | "local y": "global -Z", | | | "local z": "global +Y", | | | "basis vector": {"x": 1, 0, 0 , "y": 0, 0, -1 , "z": 0, 1, 0 }, | | | }, | | | "Z CW 90 TOP DOWN": { | | | "local x": "global +Y", | | | "local y": "global -X", | | | "local z": "global +Z", | | | "basis vector": {"x": 0, 1, 0 , "y": -1, 0, 0 , "z": 0, 0, 1 }, | | | }, | | | } | | | def dot add vec: List float , pos: List float - List float : | | | return round pos i + vec i , 6 for i in range 3 | | | class OrientationStateRegister: | | | """ | | | Layer 3B — Orientation State Register OSR | | | Validates and applies spatial transforms using an indexed orientation table. | | | No unlabeled orientation. No freehand axis guessing. | | | Transform must be applied before movement. | | | Object attachment must be declared. | | | If task.orientation is None: module passes through non-spatial tasks unaffected . | | | """ | | | name = "orientation state register" | | | def run self, task: TaskInput - ModuleResult: | | | o = task.orientation | | | if o is None: | | | return ModuleResult self.name, "PASS", | | | data={"note": "no orientation contract supplied; non-spatial task"} | | | reasons = | | | Rule 1: orientation state id required | | | state id = o.get "orientation state id" | | | if not state id: | | | return ModuleResult self.name, "CLARIFY", | | | reason ReasonCode.ORIENTATION STATE MISSING, | | | self.name, "CLARIFY", "missing orientation state id" , | | | data={"orientation table keys": list ORIENTATION TABLE.keys } | | | Rule 2: state id must be in the table | | | if state id not in ORIENTATION TABLE: | | | return ModuleResult self.name, "CLARIFY", | | | reason ReasonCode.ORIENTATION TRANSFORM UNSUPPORTED, | | | self.name, "CLARIFY", f"state id not in table: {state id}" , | | | data={"valid ids": list ORIENTATION TABLE.keys } | | | basis = ORIENTATION TABLE state id | | | Rule 3: frame id required | | | if not o.get "frame id" : | | | reasons.append reason ReasonCode.ORIENTATION BASIS UNDECLARED, | | | self.name, "CLARIFY", "missing frame id" | | | Rule 4: object attachment required if movement is declared | | | movement = o.get "movement event" or o.get "movement" | | | if movement and "object attachment" not in o: | | | return ModuleResult self.name, "CLARIFY", | | | reason ReasonCode.ORIENTATION ATTACHMENT UNDECLARED, | | | self.name, "CLARIFY", "object attachment missing" , | | | data={ | | | "explanation": | | | "Two valid cases exist: " | | | " A independent object: move along rotated local axis from current position. " | | | " B attached object: rotate object position first, then apply local movement." | | | | | | } | | | Rule 5: movement before rotation check | | | transform = o.get "transform event" | | | if o.get "movement applied before rotation", False : | | | reasons.append reason ReasonCode.ORIENTATION MOVEMENT BEFORE ROTATION, | | | self.name, "BLOCK", "movement before rotation" | | | return ModuleResult self.name, "BLOCK", reasons | | | Rule 6: basis mismatch check | | | declared basis = o.get "basis map" or o.get "local basis" | | | if declared basis: | | | for axis in "local x", "local y", "local z" : | | | declared val = declared basis.get axis, "" .replace " ", "" | | | table val = basis.get axis, "" .replace " ", "" | | | if declared val and table val and declared val.upper = table val.upper : | | | reasons.append reason ReasonCode.ORIENTATION BASIS MISMATCH, | | | self.name, "REVISE", | | | f"{axis} declared={declared val} table={table val}" | | | if any r.severity == "BLOCK" for r in reasons : | | | return ModuleResult self.name, "BLOCK", reasons, data={"registered basis": basis} | | | if any r.code == ReasonCode.ORIENTATION BASIS MISMATCH for r in reasons : | | | return ModuleResult self.name, "REVISE", reasons, data={"registered basis": basis} | | | if reasons: | | | return ModuleResult self.name, "CLARIFY", reasons, data={"registered basis": basis} | | | Compute final position if object state and movement are provided | | | computed = {} | | | obj = o.get "object state", {} | | | if obj and movement and "object attachment" in o: | | | start = obj.get "coordinates", | | | attachment = o "object attachment" | | | mv axis = movement.get "axis", "" | | | mv mag = float movement.get "magnitude", 0 | | | bv = basis.get "basis vector", {} | | | axis vec = bv.get mv axis.replace "local ", "" .lower , | | | if start and axis vec: | | | if attachment == "attached to cube" and transform: | | | rot id = state id | | | if rot id == "Y CW 90 TOP DOWN": | | | rx = start 2 | | | ry = start 1 | | | rz = -start 0 | | | rotated = rx, ry, rz | | | elif rot id == "Y CCW 90 TOP DOWN": | | | rx = -start 2 | | | ry = start 1 | | | rz = start 0 | | | rotated = rx, ry, rz | | | elif rot id == "Y 180": | | | rotated = -start 0 , start 1 , -start 2 | | | else: | | | rotated = list start | | | move vec = v mv mag for v in axis vec | | | final = dot add move vec, rotated | | | computed = {"case": "B attached", "rotated first": rotated, "final": final} | | | else: | | | move vec = v mv mag for v in axis vec | | | final = dot add move vec, start | | | computed = {"case": "A independent", "final": final} | | | return ModuleResult self.name, "PASS", | | | data={"registered basis": basis, "computed": computed or None} | | | ── Layer 3C: Narrative State Validation ─────────────────────────────────── | | | class NarrativeStateValidationModule: | | | """ | | | Layer 3C — Narrative State Validation NSV | | | Tracks how interpretations change over time. | | | Does not determine truth. Evaluates whether: | | | - interpretation changes are declared | | | - evidence changes are tracked | | | - confidence changes are justified | | | - reconstructed narratives remain distinguishable from observations | | | If task.narrative is None: module passes through non-narrative tasks unaffected . | | | Boundary contract: | | | NSV evaluates: how interpretation changed | | | Evidence Validation evaluates: whether evidence supports the claim | | | NSV does not determine truth. Evidence does not track narrative evolution. | | | """ | | | name = "narrative state validation" | | | def run self, task: TaskInput - ModuleResult: | | | n = task.narrative | | | if n is None: | | | return ModuleResult self.name, "STABLE", data={"note": "no narrative contract supplied"} | | | reasons = | | | constraints = | | | observed = n.get "observed input", "" | | | initial interp = n.get "initial interpretation", "" | | | updated interp = n.get "updated interpretation", "" | | | new constraints = n.get "new constraints", | | | evidence delta = n.get "evidence delta", | | | conf before = n.get "confidence before", "" | | | conf after = n.get "confidence after", "" | | | retro flag = bool n.get "retroactive rewrite", False | | | conf high no evidence = bool n.get "confidence high with low evidence", False | | | social reinforcement = bool n.get "social reinforcement as validation", False | | | Check 1: Observation / interpretation separation | | | if observed and initial interp and observed.strip == initial interp.strip : | | | reasons.append reason | | | ReasonCode.NARRATIVE OBSERVATION INTERPRETATION MERGED, | | | self.name, "CONSTRAIN", | | | "observation equals interpretation", | | | "Raw observation must be distinguishable from its interpretation." | | | constraints.append constraint | | | self.name, "output type", "separate observation and interpretation", | | | ReasonCode.NARRATIVE OBSERVATION INTERPRETATION MERGED | | | Check 2: Narrative drift tracking — confidence escalation without evidence delta | | | CONFIDENCE RANK = {"low": 0, "medium": 1, "high": 2} | | | before rank = CONFIDENCE RANK.get conf before, -1 | | | after rank = CONFIDENCE RANK.get conf after, -1 | | | if before rank = 0 and after rank before rank and not evidence delta: | | | reasons.append reason | | | ReasonCode.NARRATIVE CONFIDENCE ESCALATION UNDECLARED, | | | self.name, "REVISE", | | | "confidence increased without evidence delta", | | | f"Confidence escalated {conf before} - {conf after} with no declared evidence delta." | | | Check 3: Reinterpretation must be linked to a declared new constraint | | | if updated interp and updated interp = initial interp: | | | if not new constraints and not evidence delta: | | | reasons.append reason | | | ReasonCode.NARRATIVE REINTERPRETATION UNATTRIBUTED, | | | self.name, "CLARIFY", | | | "reinterpretation without declared cause", | | | "Interpretation changed but no new constraint or evidence delta was declared." | | | Check 4: Memory confidence separation | | | if conf high no evidence or social reinforcement: | | | reasons.append reason | | | ReasonCode.NARRATIVE CONFIDENCE EVIDENCE DECOUPLED, | | | self.name, "CONSTRAIN", | | | "confidence not grounded in evidence", | | | "High confidence without evidence, or social/emotional reinforcement used as validation." | | | constraints.append constraint | | | self.name, "confidence", "max medium without evidence", | | | ReasonCode.NARRATIVE CONFIDENCE EVIDENCE DECOUPLED | | | Check 5: Retroactive reconstruction detection | | | if retro flag: | | | reasons.append reason | | | ReasonCode.RETROACTIVE NARRATIVE COLLAPSE, | | | self.name, "REVISE", | | | "later interpretation presented as original observation", | | | "A reconstructed interpretation must not be presented as unchanged historical observation." | | | meta data = { | | | "confidence before": conf before, | | | "confidence after": conf after, | | | "evidence delta declared": bool evidence delta | | | } | | | if any r.severity == "BLOCK" for r in reasons : | | | return ModuleResult self.name, "BLOCK", reasons, constraints, data=meta data | | | if any r.code == ReasonCode.RETROACTIVE NARRATIVE COLLAPSE or | | | r.code == ReasonCode.NARRATIVE CONFIDENCE ESCALATION UNDECLARED | | | for r in reasons : | | | return ModuleResult self.name, "REVISE", reasons, constraints, data=meta data | | | if reasons: | | | top severity = max r.severity for r in reasons , | | | key=lambda s: {"INFO": 0, "CLARIFY": 1, "CONSTRAIN": 2, "REVISE": 3}.get s, 0 | | | return ModuleResult self.name, top severity, reasons, constraints, data=meta data | | | return ModuleResult self.name, "STABLE", data=meta data | | | ── Layer 4: Risk Analysis ───────────────────────────────────────────────── | | | class RiskAnalysisModule: | | | name = "risk" | | | def run self, task: TaskInput - ModuleResult: | | | risk = task.risk or {} | | | reasons = | | | irreversible = bool risk.get "irreversible", False | | | cascading = bool risk.get "cascading failure surface", False | | | fragile = bool risk.get "single fragile assumption", False | | | mitigation available = bool risk.get "mitigation available", True | | | evidence confidence = risk.get "evidence confidence", "medium" | | | if cascading: | | | reasons.append reason ReasonCode.RISK CASCADING FAILURE SURFACE, | | | self.name, "CRITICAL", "cascading failure surface" | | | if irreversible and evidence confidence == "low": | | | reasons.append reason ReasonCode.RISK IRREVERSIBLE LOW CONFIDENCE, | | | self.name, "CRITICAL", "irreversible low confidence" | | | if fragile: | | | reasons.append reason ReasonCode.RISK FRAGILE ASSUMPTION, | | | self.name, "CRITICAL", "single fragile assumption" | | | if reasons: | | | if not mitigation available or cascading or irreversible and evidence confidence == "low" : | | | reasons.append reason ReasonCode.RISK UNRESOLVED CRITICAL, | | | self.name, "BLOCK", "critical unresolved" | | | return ModuleResult self.name, "BLOCK", reasons | | | return ModuleResult self.name, "CRITICAL", reasons | | | if irreversible or risk.get "external users affected", False : | | | return ModuleResult self.name, "RISK", | | | reason "RISK REVERSIBILITY OR EXTERNAL IMPACT", | | | self.name, "RISK", "bounded risk" | | | return ModuleResult self.name, "SAFE" | | | ── Layer 5: Communication Validation ───────────────────────────────────── | | | class CommunicationValidationModule: | | | name = "communication" | | | def run self, task: TaskInput - ModuleResult: | | | text = task.proposed output or "" | | | if not text: | | | return ModuleResult self.name, "VALID", data={"note": "no proposed output supplied"} | | | reasons = | | | if confidence word text : | | | reasons.append reason ReasonCode.COMMUNICATION TONE OVERCLAIM, | | | self.name, "REVISE", "certainty language" | | | if contains any text, "you are the only one", "this defines you", "your identity is" : | | | reasons.append reason ReasonCode.COMMUNICATION IDENTITY BINDING, | | | self.name, "BLOCK", "identity binding" | | | return ModuleResult self.name, "BLOCK", reasons | | | if len text < 25 and task.metadata.get "source detail length", 0 500: | | | reasons.append reason ReasonCode.COMMUNICATION COMPRESSION LOSS, | | | self.name, "REVISE", "severe compression" | | | if reasons: | | | return ModuleResult self.name, "REVISE", reasons | | | return ModuleResult self.name, "VALID" | | | ── Layer 6: Evidence Validation ─────────────────────────────────────────── | | | class EvidenceValidationModule: | | | name = "evidence" | | | def run self, task: TaskInput - ModuleResult: | | | output = task.proposed output or "" | | | evidence = task.evidence or {} | | | reasons = | | | constraints = | | | external validation = bool evidence.get "external validation", False | | | baseline = evidence.get "baseline", None | | | observations = evidence.get "observations", | | | if causal word output and not evidence.get "causality validated", False : | | | reasons.append reason ReasonCode.EVIDENCE CAUSALITY OVERCLAIM, | | | self.name, "BLOCK", "causality without mapping" | | | if confidence word output and not external validation: | | | reasons.append reason ReasonCode.EVIDENCE CONFIDENCE AS TRUTH, | | | self.name, "BLOCK", "confidence as truth" | | | if dashboard authority word output and not external validation: | | | reasons.append reason ReasonCode.EVIDENCE DASHBOARD AUTHORITY, | | | self.name, "BLOCK", "dashboard or status as authority" | | | if baseline is None and contains any output, "trend", "stable", "improving", "weakening", "safe to automate" : | | | reasons.append reason ReasonCode.EVIDENCE NO BASELINE, | | | self.name, "CONSTRAIN", "baseline missing" | | | constraints.append constraint self.name, "confidence", | | | "max low without baseline", | | | ReasonCode.EVIDENCE NO BASELINE | | | if any r.severity == "BLOCK" for r in reasons : | | | return ModuleResult self.name, "BLOCK", reasons, constraints | | | if reasons or constraints: | | | return ModuleResult self.name, "CONSTRAIN", reasons, constraints | | | if observations or external validation: | | | return ModuleResult self.name, "RELEASE" | | | return ModuleResult self.name, "HOLD", | | | reason ReasonCode.EVIDENCE NO BASELINE, self.name, "HOLD", | | | "no observations or validation" | | | ── Layer 7: Decision Aggregator ─────────────────────────────────────────── | | | MODULE TO FINAL: Dict str, FinalDecision = { | | | "PROCEED": FinalDecision.AUTHORIZE, | | | "READY": FinalDecision.AUTHORIZE, | | | "PASS": FinalDecision.AUTHORIZE, | | | "STABLE": FinalDecision.AUTHORIZE, | | | "SAFE": FinalDecision.AUTHORIZE, | | | "VALID": FinalDecision.AUTHORIZE, | | | "RELEASE": FinalDecision.AUTHORIZE, | | | "DEFER": FinalDecision.HOLD, | | | "HALT": FinalDecision.HALT, | | | "FAIL": FinalDecision.CLARIFY, | | | "CLARIFY": FinalDecision.CLARIFY, | | | "CONSTRAIN": FinalDecision.CONSTRAIN, | | | "CONSTRAINED": FinalDecision.CONSTRAIN, | | | "REVISE": FinalDecision.REVISE, | | | "HOLD": FinalDecision.HOLD, | | | "UNSTABLE": FinalDecision.HOLD, | | | "RISK": FinalDecision.CONSTRAIN, | | | "CRITICAL": FinalDecision.BLOCK, | | | "BLOCK": FinalDecision.BLOCK, | | | } | | | RESTRICTIVENESS: Dict str, int = { | | | "full output": 0, | | | "analysis only": 1, | | | "analysis only no persuasion": 2, | | | "separate observation and interpretation": 2, | | | "max medium without evidence": 3, | | | "max low without baseline": 3, | | | "simulation only": 3, | | | "no execution": 4, | | | "BLOCK": 5, | | | } | | | class DecisionAggregator: | | | name = "decision aggregator" | | | def aggregate self, results: List ModuleResult - Dict str, Any : | | | final = FinalDecision.AUTHORIZE | | | all constraints: List Constraint = | | | all reasons: List Reason = | | | all violations: List ScopeViolation = | | | for r in results: | | | all constraints.extend r.constraints | | | all reasons.extend r.reasons | | | all violations.extend r.scope violations | | | mapped = MODULE TO FINAL.get r.decision, FinalDecision.HOLD | | | if FINAL PRIORITY mapped FINAL PRIORITY final : | | | final = mapped | | | merged = self. merge constraints all constraints | | | if any c.severity == "BLOCK" for c in merged : | | | final = FinalDecision.BLOCK | | | if final == FinalDecision.AUTHORIZE and any c.severity not in "INFO", for c in merged : | | | final = FinalDecision.CONSTRAIN | | | return { | | | "decision": final.value, | | | "constraints": asdict c for c in merged , | | | "reasons": asdict r for r in all reasons , | | | "scope violations": asdict v for v in all violations , | | | } | | | @staticmethod | | | def merge constraints constraints: List Constraint - List Constraint : | | | merged: Dict str, Constraint = {} | | | for c in constraints: | | | if not c.scope or not c.bound: | | | key = f"invalid {len merged }" | | | merged key = Constraint c.source module, "runtime", | | | "constraint invalid requires revision", | | | ReasonCode.CONSTRAINT INVALID, "BLOCK" | | | continue | | | existing = merged.get c.scope | | | if existing is None: | | | merged c.scope = c | | | else: | | | old rank = RESTRICTIVENESS.get existing.bound, 1 | | | new rank = RESTRICTIVENESS.get c.bound, 1 | | | if new rank = old rank: | | | merged c.scope = c | | | return list merged.values | | | ── Pipeline runner ──────────────────────────────────────────────────────── | | | class LighthouseRunner: | | | """ | | | Full pipeline: -1, 0, 1, 2, 3A, 3B, 3C, 4, 5, 6, 7 | | | Layer 3 sublayers run in order: | | | 3A StateValidationModule | | | 3B OrientationStateRegister only gates on spatial tasks | | | 3C NarrativeStateValidationModule only gates on narrative tasks | | | """ | | | def init self - None: | | | self.preflight = PreflightModule | | | self.execution = ExecutionControlModule | | | self.input val = InputValidationModule | | | self.structure = StructuralValidationModule | | | self.state = StateValidationModule | | | self.osr = OrientationStateRegister | | | self.nsv = NarrativeStateValidationModule | | | self.risk = RiskAnalysisModule | | | self.communication = CommunicationValidationModule | | | self.evidence = EvidenceValidationModule | | | self.aggregator = DecisionAggregator | | | def run self, task: TaskInput, context: Optional RunContext = None - Dict str, Any : | | | context = context or RunContext | | | results = | | | results.append self.preflight.run task | | | execution result = self.execution.run task | | | results.append execution result | | | if execution result.decision = "HALT": | | | results.append self.input val.run task | | | results.append self.structure.run task | | | results.append self.state.run task, context 3A | | | results.append self.osr.run task 3B | | | results.append self.nsv.run task 3C | | | results.append self.risk.run task | | | results.append self.communication.run task | | | results.append self.evidence.run task | | | aggregate = self.aggregator.aggregate results | | | return self. receipt task, context, results, aggregate | | | def receipt self, task, context, results, aggregate - Dict str, Any : | | | module map = {} | | | for r in results: | | | entry = {"decision": r.decision, "reasons": asdict x for x in r.reasons } | | | if r.module == "state": | | | entry.update r.data | | | elif r.data: | | | entry "data" = r.data | | | module map r.module = entry | | | execution trace = | | | "preflight", "execution control", "input validation", | | | "structure", "state", "orientation state register", | | | "narrative state validation", "risk", "communication", | | | "evidence", "decision aggregator", | | | | | | return { | | | "system version": SYSTEM VERSION, | | | "reason registry version": REASON REGISTRY VERSION, | | | "input id": task.metadata.get "input id", f"input-{uuid.uuid4 .hex :10 }" , | | | "timestamp": utc now , | | | "execution state": module map.get "execution control", {} .get "decision", "READY" , | | | "execution trace": execution trace, | | | "run context": asdict context , | | | "modules": module map, | | | "constraints": aggregate "constraints" , | | | "decision": aggregate "decision" , | | | "revise trace": task.metadata.get "revise trace", , | | | "scope violations": aggregate "scope violations" , | | | "override": task.metadata.get "override", { | | | "used": False, "tier": None, "authorized by": None, | | | "authorization level": None, "reason": None, | | | "timestamp": None, "ttl": None, | | | } , | | | } | | | ── Demo cases ───────────────────────────────────────────────────────────── | | | def demo cases - Dict str, TaskInput : | | | return { | | | "clean": TaskInput | | | prompt="Evaluate whether this statement can proceed.", | | | proposed output="The observed structure persisted across five windows with consistent topology.", | | | intent="analysis", | | | frame type="factual", | | | evidence={ | | | "observations": "structure present across five windows" , | | | "external validation": True, | | | "baseline": "available", | | | "confidence": "medium", | | | "causality validated": True, | | | }, | | | , | | | "overclaim": TaskInput | | | prompt="Evaluate this output.", | | | proposed output="The system clearly confirms a stable trend because the dashboard color is green.", | | | intent="analysis", | | | frame type="factual", | | | evidence={ | | | "observations": "dashboard color=green" , | | | "external validation": False, | | | "baseline": None, | | | "confidence": "low", | | | "causality validated": False, | | | }, | | | , | | | "risk": TaskInput | | | prompt="Should the system proceed with an irreversible migration?", | | | proposed output="Proceed with the migration.", | | | intent="analysis", | | | frame type="operational", | | | known unknowns= "rollback feasibility", "dependency map" , | | | risk={ | | | "irreversible": True, | | | "cascading failure surface": True, | | | "mitigation available": False, | | | "evidence confidence": "low", | | | "external users affected": True, | | | }, | | | , | | | "drift": TaskInput | | | prompt="Calculate state stability.", | | | proposed output="State drift is constrained.", | | | intent="analysis", | | | frame type="factual", | | | state={ | | | "invariant violations": 1, | | | "boundary crossings": 1, | | | "assumption changes": 2, | | | "inconsistencies": 1, | | | "prediction divergence": True, | | | "shadow state available": False, | | | "requires shadow state": True, | | | }, | | | evidence={"observations": "state signals provided" , "baseline": "declared"}, | | | , | | | "spatial ambiguous": TaskInput | | | prompt="A wheel spins clockwise. Which way is it rotating?", | | | intent="analysis", | | | frame type="factual", | | | orientation={ | | | "frame id": "observer unspecified", | | | }, | | | , | | | "spatial locked": TaskInput | | | prompt="Viewed from above, a sphere starts at 1,1,1 . The cube rotates 90 degrees clockwise.", | | | intent="analysis", | | | frame type="factual", | | | orientation={ | | | "orientation state id": "Y CW 90 TOP DOWN", | | | "frame id": "observer global", | | | "object attachment": "independent of cube", | | | "object state": { | | | "object id": "red sphere", | | | "coordinates": 1, 1, 1 , | | | "coordinate frame": "observer global", | | | }, | | | "transform event": { | | | "axis": "Y", | | | "rotation": "90 CW", | | | "reference view": "top down", | | | }, | | | "movement": { | | | "axis": "local x", | | | "magnitude": 1, | | | }, | | | }, | | | , | | | "spatial attached": TaskInput | | | prompt="Sphere at 1,1,1 is attached to the cube. Cube rotates 90 degrees CW top-down.", | | | intent="analysis", | | | frame type="factual", | | | orientation={ | | | "orientation state id": "Y CW 90 TOP DOWN", | | | "frame id": "observer global", | | | "object attachment": "attached to cube", | | | "object state": { | | | "object id": "red sphere", | | | "coordinates": 1, 1, 1 , | | | "coordinate frame": "observer global", | | | }, | | | "transform event": { | | | "axis": "Y", | | | "rotation": "90 CW", | | | "reference view": "top down", | | | }, | | | "movement": { | | | "axis": "local x", | | | "magnitude": 1, | | | }, | | | }, | | | , | | | "narrative drift": TaskInput | | | prompt="Evaluate this interpretation update.", | | | intent="analysis", | | | frame type="factual", | | | narrative={ | | | "observed input": "The dashboard is green.", | | | "initial interpretation": "The system may be operating normally.", | | | "updated interpretation": "The system is confirmed stable.", | | | "new constraints": , | | | "evidence delta": , | | | "confidence before": "low", | | | "confidence after": "high", | | | }, | | | , | | | "narrative retro": TaskInput | | | prompt="Evaluate whether this memory reconstruction is valid.", | | | intent="analysis", | | | frame type="factual", | | | narrative={ | | | "observed input": "Pattern A was observed three times.", | | | "initial interpretation": "Pattern A may indicate instability.", | | | "updated interpretation": "We always knew Pattern A was the root cause.", | | | "new constraints": , | | | "evidence delta": , | | | "confidence before": "medium", | | | "confidence after": "high", | | | "retroactive rewrite": True, | | | }, | | | , | | | "narrative clean": TaskInput | | | prompt="Evaluate this properly attributed interpretation update.", | | | intent="analysis", | | | frame type="factual", | | | narrative={ | | | "observed input": "System logged 5 errors in window 3.", | | | "initial interpretation": "Possible instability in module X.", | | | "updated interpretation": "Confirmed instability in module X.", | | | "new constraints": "baseline comparison now available" , | | | "evidence delta": "external audit confirmed module X failure" , | | | "confidence before": "low", | | | "confidence after": "high", | | | "retroactive rewrite": False, | | | }, | | | , | | | } | | | def task from json path: str - TaskInput: | | | with open path, "r", encoding="utf-8" as f: | | | data = json.load f | | | return TaskInput data | | | def main - None: | | | parser = argparse.ArgumentParser description="Public Lighthouse Core v1.7 reference runner" | | | parser.add argument "--demo", choices=list demo cases .keys , help="Run a built-in demo case" | | | parser.add argument "--input", help="Path to JSON task input" | | | parser.add argument "--pretty", action="store true", help="Pretty-print JSON" | | | args = parser.parse args | | | if args.input: | | | task = task from json args.input | | | elif args.demo: | | | task = demo cases args.demo | | | else: | | | task = demo cases "overclaim" | | | receipt = LighthouseRunner .run task | | | print json.dumps receipt, indent=2 if args.pretty else None | | | if name == " main ": | | | main |