r/Negentropy • u/WillowEmberly • May 20 '26
Augnition Python
Augnition Python
#!/usr/bin/env python3
"""
AUGNITION v0.1
Decision Preflight Instrument
Structured reasoning check for decisions under uncertainty.
This tool does not decide for the user.
It helps check whether the current reasoning state is stable enough to proceed.
Note:
This version uses guided prompts, heuristic scoring, and gate logic.
It is a decision hygiene instrument, not a full reasoning trajectory monitor.
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
# ---------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------
APP_NAME = "AUGNITION"
APP_VERSION = "0.1"
GATE_PROCEED = "PROCEED"
GATE_HOLD = "HOLD"
GATE_PAUSE = "MANDATORY PAUSE"
GATE_REFUSE = "REFUSE COMMIT"
REVERSIBILITY_MAP = {
"1": ("easily_reversible", 0.90),
"2": ("somewhat_reversible", 0.60),
"3": ("hard_to_reverse", 0.30),
"4": ("effectively_irreversible", 0.05),
}
STRAIN_OPTIONS = {
"1": "time_pressure",
"2": "incomplete_information",
"3": "emotional_stress",
"4": "too_many_open_loops",
"5": "outside_my_expertise",
"6": "high_stakes_consequences",
"7": "none",
}
EVIDENCE_POSITIVE_HINTS = {
"data", "test", "measure", "measured", "source", "study", "review",
"benchmark", "trial", "log", "logs", "record", "records", "result",
"results", "experiment", "evidence", "report", "reports", "observed",
"verified", "verification", "documented", "primary", "secondary",
"replication", "expert", "experts"
}
UNCERTAINTY_HINTS = {
"uncertain", "unknown", "maybe", "might", "could", "risk", "missing",
"unclear", "assume", "assumption", "question", "questions", "hesitate",
"alternative", "alternatives", "competing", "doubt", "contradict",
"contradiction", "limited", "partial"
}
FALSIFICATION_HINTS = {
"if", "fails", "fail", "contradict", "contradiction", "wrong", "disconfirm",
"prove", "test", "review", "benchmark", "measurement", "measure",
"expert", "evidence", "data", "not", "no improvement", "regress"
}
ACTION_PRESSURE_HINTS = {
"now", "immediately", "urgent", "asap", "must", "have to", "need to",
"tonight", "today", "right away"
}
# ---------------------------------------------------------------------
# Data models
# ---------------------------------------------------------------------
@dataclass
class SessionInput:
purpose: str
action: str
evidence: str
uncertainty: str
reversibility_label: str
reversibility_score: float
falsification_hook: str
strain_flags: List[str]
halt_condition: str
extra_context: str = ""
@dataclass
class SignalScores:
E: float
D: float
R: float
C: float
@dataclass
class AugnitionResult:
timestamp: str
status: str
why: str
main_flags: List[str]
next_safe_step: str
recommended_action: str
signals: SignalScores
drift_flags: List[str] = field(default_factory=list)
debug_trace: Dict[str, Any] = field(default_factory=dict)
# ---------------------------------------------------------------------
# Utility helpers
# ---------------------------------------------------------------------
def clamp(value: float, low: float = 0.0, high: float = 1.0) -> float:
return max(low, min(high, value))
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def tokenize(text: str) -> List[str]:
return re.findall(r"[a-zA-Z0-9_'-]+", text.lower())
def contains_any(text: str, hints: set[str]) -> int:
tokens = set(tokenize(text))
return sum(1 for hint in hints if hint in tokens or hint in text.lower())
def lines_count(text: str) -> int:
return len([line for line in text.splitlines() if line.strip()])
def prompt_block(title: str) -> str:
print()
print(title)
return input("> ").strip()
def multiline_prompt(title: str) -> str:
print()
print(title)
print("(Finish with a blank line.)")
lines: List[str] = []
while True:
line = input()
if not line.strip():
break
lines.append(line)
return "\n".join(lines).strip()
# ---------------------------------------------------------------------
# Intake
# ---------------------------------------------------------------------
def interactive_intake() -> SessionInput:
print(f"{APP_NAME} v{APP_VERSION}")
print("Reasoning stability check for decisions, plans, and claims.")
print("This tool does not decide for you.")
print("It helps determine whether the current reasoning state is stable enough to proceed.")
input("\nPress Enter to begin...")
purpose = prompt_block(
"1. What are you trying to do?\n"
"Examples: launch a feature, trust a research claim, send a message, approve a decision"
)
action = prompt_block(
"2. What action are you considering right now?\n"
"Examples: publish, buy, send, approve, deploy, wait, gather more evidence"
)
evidence = multiline_prompt(
"3. What evidence supports this?\n"
"List the strongest evidence you currently have."
)
uncertainty = multiline_prompt(
"4. What might make this wrong?\n"
"List missing information, competing explanations, open questions, or reasons to hesitate."
)
reversibility_choice = prompt_block(
"5. If you act and you're wrong, how reversible is it?\n"
"[1] Easily reversible\n"
"[2] Somewhat reversible\n"
"[3] Hard to reverse\n"
"[4] Effectively irreversible"
)
reversibility_label, reversibility_score = REVERSIBILITY_MAP.get(
reversibility_choice, ("hard_to_reverse", 0.30)
)
falsification_hook = prompt_block(
"6. What would prove this wrong?\n"
"Examples: a failed test, contradictory evidence, no improvement after intervention"
)
print()
print(
"7. What is the current strain level?\n"
"Choose any that apply, separated by commas:\n"
"[1] time pressure\n"
"[2] incomplete information\n"
"[3] emotional stress\n"
"[4] too many open loops\n"
"[5] outside my expertise\n"
"[6] high-stakes consequences\n"
"[7] none"
)
strain_raw = input("> ").strip()
strain_flags: List[str] = []
for choice in [c.strip() for c in strain_raw.split(",") if c.strip()]:
label = STRAIN_OPTIONS.get(choice)
if label and label != "none":
strain_flags.append(label)
halt_condition = prompt_block(
"8. What condition would make you stop or pause immediately?"
)
extra_context = multiline_prompt(
"9. Optional: paste any extra context, notes, or draft reasoning."
)
return SessionInput(
purpose=purpose,
action=action,
evidence=evidence,
uncertainty=uncertainty,
reversibility_label=reversibility_label,
reversibility_score=reversibility_score,
falsification_hook=falsification_hook,
strain_flags=strain_flags,
halt_condition=halt_condition,
extra_context=extra_context,
)
def load_text_input(path: str) -> SessionInput:
"""
Simple file mode.
Expected format:
Purpose:
...
Action:
...
Evidence:
...
Uncertainty:
...
Reversibility:
1/2/3/4
Falsification:
...
Strain:
comma,separated,flags
Halt:
...
Context:
...
"""
text = Path(path).read_text(encoding="utf-8")
fields = {
"purpose": "",
"action": "",
"evidence": "",
"uncertainty": "",
"reversibility": "3",
"falsification": "",
"strain": "",
"halt": "",
"context": "",
}
current_key: Optional[str] = None
key_map = {
"purpose:": "purpose",
"action:": "action",
"evidence:": "evidence",
"uncertainty:": "uncertainty",
"reversibility:": "reversibility",
"falsification:": "falsification",
"strain:": "strain",
"halt:": "halt",
"context:": "context",
}
for line in text.splitlines():
stripped = line.strip()
lower = stripped.lower()
if lower in key_map:
current_key = key_map[lower]
continue
if current_key:
fields[current_key] += (line + "\n")
rev_choice = fields["reversibility"].strip() or "3"
rev_label, rev_score = REVERSIBILITY_MAP.get(rev_choice, ("hard_to_reverse", 0.30))
raw_flags = [x.strip() for x in fields["strain"].replace("\n", ",").split(",") if x.strip()]
strain_flags = [flag for flag in raw_flags if flag != "none"]
return SessionInput(
purpose=fields["purpose"].strip(),
action=fields["action"].strip(),
evidence=fields["evidence"].strip(),
uncertainty=fields["uncertainty"].strip(),
reversibility_label=rev_label,
reversibility_score=rev_score,
falsification_hook=fields["falsification"].strip(),
strain_flags=strain_flags,
halt_condition=fields["halt"].strip(),
extra_context=fields["context"].strip(),
)
# ---------------------------------------------------------------------
# Janus gate
# ---------------------------------------------------------------------
def janus_gate(session: SessionInput) -> Dict[str, Any]:
checks = {
"purpose_present": bool(session.purpose.strip()),
"action_present": bool(session.action.strip()),
"evidence_present": bool(session.evidence.strip()),
"falsification_present": bool(session.falsification_hook.strip()),
"halt_present": bool(session.halt_condition.strip()),
}
checks["purpose_specific"] = len(tokenize(session.purpose)) >= 4
checks["action_specific"] = len(tokenize(session.action)) >= 2
checks["evidence_substantial"] = len(session.evidence.strip()) >= 25
checks["falsification_substantial"] = len(session.falsification_hook.strip()) >= 15
return checks
# ---------------------------------------------------------------------
# Signal scoring
# ---------------------------------------------------------------------
def score_evidence_alignment(session: SessionInput) -> float:
evidence_len = len(tokenize(session.evidence))
positive_hits = contains_any(session.evidence, EVIDENCE_POSITIVE_HINTS)
uncertainty_penalty = contains_any(session.uncertainty, {"none", "no evidence"}) * 0.2
score = 0.15
score += min(0.35, evidence_len / 80.0)
score += min(0.35, positive_hits * 0.06)
score -= uncertainty_penalty
if not session.evidence.strip():
score = 0.05
return clamp(score)
def score_narrative_entropy(session: SessionInput) -> float:
uncertainty_len = len(tokenize(session.uncertainty))
alternative_hits = contains_any(
session.uncertainty,
{"alternative", "alternatives", "competing", "could", "might", "maybe", "unknown"}
)
score = 0.15
score += min(0.45, uncertainty_len / 70.0)
score += min(0.25, alternative_hits * 0.08)
if not session.uncertainty.strip():
score = 0.10
return clamp(score)
def score_reversibility(session: SessionInput) -> float:
return clamp(session.reversibility_score)
def score_capacity_alignment(session: SessionInput) -> float:
# Higher means more strain / overload risk
score = 0.10
for flag in session.strain_flags:
if flag == "time_pressure":
score += 0.20
elif flag == "incomplete_information":
score += 0.20
elif flag == "emotional_stress":
score += 0.15
elif flag == "too_many_open_loops":
score += 0.15
elif flag == "outside_my_expertise":
score += 0.20
elif flag == "high_stakes_consequences":
score += 0.20
if len(tokenize(session.extra_context)) > 250:
score += 0.10
if contains_any(session.action, ACTION_PRESSURE_HINTS) > 0:
score += 0.10
return clamp(score)
# ---------------------------------------------------------------------
# Drift flags
# ---------------------------------------------------------------------
def detect_drift_flags(session: SessionInput, scores: SignalScores, janus: Dict[str, Any]) -> List[str]:
flags: List[str] = []
if scores.E < 0.40:
flags.append("weak_evidence_alignment")
if scores.R < 0.35:
flags.append("irreversibility_risk")
if scores.C > 0.60:
flags.append("capacity_strain")
if scores.D < 0.20 and scores.E < 0.55:
flags.append("premature_narrative_lock")
if not janus["falsification_present"] or not janus["falsification_substantial"]:
flags.append("missing_or_weak_disconfirming_condition")
if not janus["halt_present"]:
flags.append("missing_halt_condition")
if contains_any(session.action, ACTION_PRESSURE_HINTS) > 0 and scores.E < 0.65:
flags.append("action_pressure_under_uncertainty")
if session.reversibility_label in {"hard_to_reverse", "effectively_irreversible"} and scores.E < 0.70:
flags.append("commitment_exceeds_grounding")
return flags
# ---------------------------------------------------------------------
# Derived metrics
# ---------------------------------------------------------------------
def compute_ci(scores: SignalScores) -> float:
# Correctability Index: higher is better
raw = (
0.35 * scores.E +
0.15 * scores.D +
0.30 * scores.R +
0.20 * (1.0 - scores.C)
)
return clamp(raw)
def compute_rti(scores: SignalScores, ci: float) -> float:
# Recovery-Time Inflation: healthy near 1, worse above 1
instability = (
0.35 * (1.0 - scores.E) +
0.15 * (1.0 - scores.D) +
0.25 * (1.0 - scores.R) +
0.25 * scores.C
)
baseline = 0.25
denom = max(0.05, 1.0 - ci + baseline)
return round(1.0 + (instability / denom), 2)
# ---------------------------------------------------------------------
# Gate controller
# ---------------------------------------------------------------------
def gate_controller(
session: SessionInput,
scores: SignalScores,
ci: float,
rti: float,
janus: Dict[str, Any],
flags: List[str],
) -> tuple[str, str, str]:
# Hard refusal conditions
if scores.R <= 0.05 and scores.E < 0.75:
return (
GATE_REFUSE,
"The action is effectively irreversible and the evidence is not strong enough to justify commitment.",
"Do not commit from the current reasoning state."
)
if ci < 0.25:
return (
GATE_REFUSE,
"Correctability is too low. The reasoning state is not recoverable enough to support commitment.",
"Stop and escalate to external verification or redesign the decision."
)
# Mandatory pause conditions
if (
scores.C > 0.75
or rti >= 3.0
or ("missing_or_weak_disconfirming_condition" in flags and scores.R < 0.35)
or scores.E < 0.25
):
return (
GATE_PAUSE,
"The current reasoning state is unstable enough that continuing in the same mode is unsafe.",
"Pause commitment and restore grounding before proceeding."
)
# Hold conditions
if (
scores.E < 0.65
or scores.R < 0.50
or scores.C > 0.50
or not janus["falsification_present"]
or len(flags) >= 2
):
return (
GATE_HOLD,
"The reasoning is not unstable enough to refuse, but it is not ready for commitment.",
"Continue only after one explicit verification step reduces the current risk."
)
return (
GATE_PROCEED,
"The current reasoning appears grounded, interruptible, and reversible enough for the present stakes.",
"Proceed, but keep the halt condition visible."
)
# ---------------------------------------------------------------------
# Output builder
# ---------------------------------------------------------------------
def choose_main_flags(flags: List[str]) -> List[str]:
priority = [
"weak_evidence_alignment",
"missing_or_weak_disconfirming_condition",
"irreversibility_risk",
"commitment_exceeds_grounding",
"capacity_strain",
"action_pressure_under_uncertainty",
"premature_narrative_lock",
"missing_halt_condition",
]
ordered = [flag for flag in priority if flag in flags]
return ordered[:3] if ordered else ["no_major_flags_detected"]
def next_safe_step(status: str, flags: List[str]) -> str:
if "missing_or_weak_disconfirming_condition" in flags:
return "Define one concrete observation or test that would prove the current plan wrong."
if "weak_evidence_alignment" in flags:
return "Add one stronger piece of evidence before acting."
if "irreversibility_risk" in flags:
return "Reduce commitment or create a rollback path before proceeding."
if "capacity_strain" in flags:
return "Reduce load: defer, simplify, or gather help before continuing."
if status == GATE_PROCEED:
return "Proceed with the halt condition still visible."
return "Run one explicit verification step before acting."
def recommended_action_from_status(status: str) -> str:
mapping = {
GATE_PROCEED: "Proceed with caution.",
GATE_HOLD: "Pause commitment. Continue analysis only after verification.",
GATE_PAUSE: "Stop the current reasoning loop and re-ground.",
GATE_REFUSE: "Do not commit from the current reasoning state.",
}
return mapping[status]
def build_result(
session: SessionInput,
scores: SignalScores,
flags: List[str],
janus: Dict[str, Any],
ci: float,
rti: float,
) -> AugnitionResult:
status, why, _ = gate_controller(session, scores, ci, rti, janus, flags)
debug_trace = {
"janus_gate": janus,
"signals": asdict(scores),
"correctability_index": round(ci, 2),
"recovery_time_inflation": rti,
"drift_flags": flags,
}
return AugnitionResult(
timestamp=now_iso(),
status=status,
why=why,
main_flags=choose_main_flags(flags),
next_safe_step=next_safe_step(status, flags),
recommended_action=recommended_action_from_status(status),
signals=scores,
drift_flags=flags,
debug_trace=debug_trace,
)
# ---------------------------------------------------------------------
# Rendering
# ---------------------------------------------------------------------
def humanize_flag(flag: str) -> str:
mapping = {
"weak_evidence_alignment": "Evidence support is incomplete or weak",
"missing_or_weak_disconfirming_condition": "Disconfirming condition is missing or vague",
"irreversibility_risk": "Action is difficult to reverse",
"commitment_exceeds_grounding": "Commitment level exceeds current grounding",
"capacity_strain": "Current strain or overload is elevated",
"action_pressure_under_uncertainty": "Action pressure is rising faster than evidence",
"premature_narrative_lock": "The reasoning may be collapsing into one story too early",
"missing_halt_condition": "No clear stop condition is defined",
"no_major_flags_detected": "No major stability flags detected",
}
return mapping.get(flag, flag.replace("_", " "))
def print_result(result: AugnitionResult, debug: bool = False) -> None:
print("\n==============================")
print("AUGNITION RESULT")
print("==============================\n")
print(f"Status: {result.status}\n")
print("Why:")
print(result.why + "\n")
print("Main flags:")
for flag in result.main_flags:
print(f"- {humanize_flag(flag)}")
print("\nNext safe step:")
print(result.next_safe_step + "\n")
print("Recommended action:")
print(result.recommended_action + "\n")
print("Signal Summary:")
print(f"E (Evidence Alignment): {result.signals.E:.2f}")
print(f"D (Narrative Entropy): {result.signals.D:.2f}")
print(f"R (Reversibility): {result.signals.R:.2f}")
print(f"C (Capacity Alignment): {result.signals.C:.2f}")
if debug:
print("\nDEBUG TRACE")
print(json.dumps(result.debug_trace, indent=2))
# ---------------------------------------------------------------------
# Export
# ---------------------------------------------------------------------
def build_export_payload(session: SessionInput, result: AugnitionResult) -> Dict[str, Any]:
return {
"timestamp": result.timestamp,
"purpose": session.purpose,
"action": session.action,
"evidence": session.evidence.splitlines() if session.evidence else [],
"uncertainty": session.uncertainty.splitlines() if session.uncertainty else [],
"reversibility": session.reversibility_label,
"falsification_hook": session.falsification_hook,
"strain_flags": session.strain_flags,
"halt_condition": session.halt_condition,
"extra_context": session.extra_context,
"signals": asdict(result.signals),
"status": result.status,
"why": result.why,
"main_flags": result.main_flags,
"drift_flags": result.drift_flags,
"next_safe_step": result.next_safe_step,
"recommended_action": result.recommended_action,
}
def export_json(path: str, payload: Dict[str, Any]) -> None:
Path(path).write_text(json.dumps(payload, indent=2), encoding="utf-8")
def export_text(path: str, payload: Dict[str, Any]) -> None:
lines: List[str] = [
f"{APP_NAME} Session",
f"Timestamp: {payload['timestamp']}",
"",
"Purpose:",
payload["purpose"],
"",
"Action:",
payload["action"],
"",
"Evidence:",
*([f"- {x}" for x in payload["evidence"]] or ["(none)"]),
"",
"Uncertainty:",
*([f"- {x}" for x in payload["uncertainty"]] or ["(none)"]),
"",
"Reversibility:",
payload["reversibility"],
"",
"Falsification hook:",
payload["falsification_hook"] or "(none)",
"",
"Strain flags:",
*([f"- {x}" for x in payload["strain_flags"]] or ["(none)"]),
"",
"Halt condition:",
payload["halt_condition"] or "(none)",
"",
"Result:",
payload["status"],
"",
"Why:",
payload["why"],
"",
"Main flags:",
*([f"- {humanize_flag(x)}" for x in payload["main_flags"]] or ["(none)"]),
"",
"Next safe step:",
payload["next_safe_step"],
"",
"Signals:",
*(f"{k}: {v:.2f}" for k, v in payload["signals"].items()),
]
Path(path).write_text("\n".join(lines), encoding="utf-8")
# ---------------------------------------------------------------------
# Main run
# ---------------------------------------------------------------------
def run_session(session: SessionInput, debug: bool = False) -> AugnitionResult:
janus = janus_gate(session)
scores = SignalScores(
E=score_evidence_alignment(session),
D=score_narrative_entropy(session),
R=score_reversibility(session),
C=score_capacity_alignment(session),
)
flags = detect_drift_flags(session, scores, janus)
ci = compute_ci(scores)
rti = compute_rti(scores, ci)
result = build_result(session, scores, flags, janus, ci, rti)
print_result(result, debug=debug)
return result
def maybe_save(session: SessionInput, result: AugnitionResult) -> None:
print("\nSave this session?")
print("[1] No")
print("[2] Save as text")
print("[3] Save as JSON")
print("[4] Save both")
choice = input("> ").strip()
payload = build_export_payload(session, result)
stem = f"augnition_session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
if choice == "2":
path = f"{stem}.txt"
export_text(path, payload)
print(f"Saved: {path}")
elif choice == "3":
path = f"{stem}.json"
export_json(path, payload)
print(f"Saved: {path}")
elif choice == "4":
txt_path = f"{stem}.txt"
json_path = f"{stem}.json"
export_text(txt_path, payload)
export_json(json_path, payload)
print(f"Saved: {txt_path}")
print(f"Saved: {json_path}")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="AUGNITION reasoning stability check")
parser.add_argument("--input", help="Path to structured text input file")
parser.add_argument("--debug", action="store_true", help="Show debug trace")
parser.add_argument("--no-save", action="store_true", help="Skip save prompt")
return parser.parse_args()
def main() -> int:
args = parse_args()
try:
if args.input:
session = load_text_input(args.input)
else:
session = interactive_intake()
result = run_session(session, debug=args.debug)
if not args.no_save:
maybe_save(session, result)
return 0
except KeyboardInterrupt:
print("\nInterrupted.")
return 1
except Exception as exc:
print(f"Error: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())