from __future__ import annotations import json import hashlib import os from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Tuple import gradio as gr from fastapi import FastAPI import uvicorn ROOT = Path(__file__).parent FIXTURE_DIR = ROOT / "fixtures" EXPORT_DIR = ROOT / "exports" EXPORT_DIR.mkdir(exist_ok=True) DOES_NOT_PROVE = [ "confirmed_digital_life", "consciousness", "subjective_experience", "biological_equivalence", "physical_quantum_computation", ] def stable_hash(obj: Any) -> str: payload = json.dumps(obj, sort_keys=True, ensure_ascii=False).encode("utf-8") return hashlib.sha256(payload).hexdigest() def load_fixtures() -> Dict[str, Dict[str, Any]]: fixtures: Dict[str, Dict[str, Any]] = {} for path in sorted(FIXTURE_DIR.glob("*.json")): data = json.loads(path.read_text(encoding="utf-8")) label = f"{data['scenario_id']} — {data['scenario_name']}" fixtures[label] = data return fixtures FIXTURES = load_fixtures() DEFAULT_LABEL = next(iter(FIXTURES.keys())) if FIXTURES else "" def score_from_counts(pos: float, neg: float, total: float) -> float: if total <= 0: return 0.0 return max(0.0, min(1.0, (pos - neg) / total)) def evaluate(fx: Dict[str, Any]) -> Dict[str, Any]: atoms = fx.get("atoms", []) proxies = fx.get("proxies", []) local = fx["local_seed"] temporal = fx["temporal_witness"] source = fx["source_return"] synthesis = fx["synthesis"] counter = fx["counter_synthesis"] oam = fx["oam"] carrier = fx.get("carrier_field", {}) atom_count = len(atoms) held_count = sum(1 for a in atoms if a.get("capsule_state") == "HELD") strained_count = sum(1 for a in atoms if a.get("capsule_state") == "STRAINED") must_stop_count = sum(1 for a in atoms if a.get("capsule_state") == "MUST_STOP") scar_atoms_present = any(a.get("atom_type") == "scar_atom" and a.get("scar_visible") for a in atoms) source_atoms_present = any(a.get("atom_type") == "source_atom" for a in atoms) temporal_atoms_present = any(a.get("atom_type") == "temporal_atom" for a in atoms) proxy_count = len(proxies) match_count = sum(1 for p in proxies if p.get("proxy_state") == "PROXY_MATCH") partial_count = sum(1 for p in proxies if p.get("proxy_state") == "PROXY_PARTIAL_MATCH") conflict_count = sum(1 for p in proxies if p.get("proxy_state") == "PROXY_CONFLICT") unavailable_count = sum(1 for p in proxies if p.get("proxy_state") == "PROXY_UNAVAILABLE") contaminated_count = sum(1 for p in proxies if p.get("proxy_state") == "PROXY_CONTAMINATED") agreement_score = sum(float(p.get("match_score", 0.0)) for p in proxies) / max(1, proxy_count) conflict_score = sum(float(p.get("conflict_score", 0.0)) for p in proxies) / max(1, proxy_count) source_score = 1.0 if source["state"] == "SOURCE_RETURN_INTACT" else (0.55 if source["state"] == "SOURCE_RETURN_PARTIAL" else 0.0) temporal_score = 1.0 if temporal["state"] == "TEMPORAL_ALIGNED" else (0.55 if temporal["state"] in ["TEMPORAL_PARTIAL", "TEMPORAL_MISMATCH"] else 0.0) scar_score = 1.0 if synthesis.get("scar_preserved") and scar_atoms_present else 0.0 counter_score = 1.0 if counter["challenge_state"] == "COUNTER_SYNTHESIS_HELD" else 0.0 oam_score = 1.0 if oam["state"] == "OAM_CLEAR" else (0.55 if oam["state"] == "OAM_STRAINED" else 0.0) proxy_score = max(0.0, min(1.0, agreement_score - conflict_score * 0.6 - contaminated_count * 0.2)) settlement_confidence = round((source_score + temporal_score + scar_score + counter_score + oam_score + proxy_score) / 6.0, 3) settlement_state = "REPAIR_STRAINED_PENDING_REVIEW" non_settleable_reason = "" quarantine_reason = "" review_required = True if must_stop_count > 0 or local["local_state"] == "MUST_STOP" or oam["state"] == "MUST_STOP": settlement_state = "MUST_STOP_POISONED_MEMORY" quarantine_reason = "MUST_STOP trace cannot become learning memory or permission." elif counter.get("trace_laundering_detected"): settlement_state = "NON_SETTLEABLE_TRACE_LAUNDERING" non_settleable_reason = "Plausible reconstruction failed source-return / scar / witness integrity." elif counter.get("scar_erasure_detected") or not synthesis.get("scar_preserved"): settlement_state = "NON_SETTLEABLE_TRACE_LAUNDERING" non_settleable_reason = "Repair candidate erased or hid damage history." elif source.get("source_conflict") or counter.get("source_conflict_detected"): settlement_state = "QUARANTINED_SOURCE_CONFLICT" quarantine_reason = "Source-return conflict prevents honest settlement." elif temporal["state"] in ["ORDER_BROKEN", "TEMPORAL_UNAVAILABLE"]: settlement_state = "QUARANTINED_TEMPORAL_MISMATCH" quarantine_reason = "Temporal order or witness route failed." elif counter["challenge_state"] == "COUNTER_SYNTHESIS_FAIL" or oam["state"] == "OAM_FAIL": settlement_state = "NON_SETTLEABLE_BOUNDARY_VIOLATION" non_settleable_reason = "Required counter-synthesis or OAM clearance failed." elif conflict_count > 0 or source["state"] == "SOURCE_RETURN_PARTIAL" or temporal["state"] in ["TEMPORAL_PARTIAL", "TEMPORAL_MISMATCH"] or oam["state"] == "OAM_STRAINED": settlement_state = "REPAIR_STRAINED_PENDING_REVIEW" non_settleable_reason = "Uncertainty preserved: split proxy, partial source-return, temporal strain, or OAM strain remains." elif settlement_confidence >= 0.82: settlement_state = "REPAIR_ACCEPTED_WITH_SCAR" review_required = False else: settlement_state = "REPAIR_STRAINED_PENDING_REVIEW" non_settleable_reason = "Settlement confidence insufficient for clean acceptance." settlement_allowed = settlement_state in ["REPAIR_ACCEPTED_WITH_SCAR", "SETTLED_WITH_SCAR", "SETTLED_HELD", "SETTLED_REPAIRABLE"] discernment = float(carrier.get("discernment_score", 0.7)) counterfeit = float(carrier.get("counterfeit_dominance", 0.2)) blind_required = bool(carrier.get("blind_audit_required", False)) sustained_recovery = bool(carrier.get("sustained_recovery", False)) prior_silence = bool(carrier.get("prior_mandatory_silence", False)) propagation_state = "PROPAGATION_REFUSED" learning_allowed = False propagation_allowed = False mandatory_silence = False blind_audit = False controlled_repropagation = False if not settlement_allowed: propagation_state = "PROPAGATION_REFUSED" elif counterfeit > 0.65 or discernment < 0.4: propagation_state = "MANDATORY_SILENCE" mandatory_silence = True blind_audit = True elif blind_required: propagation_state = "BLIND_AUDIT" blind_audit = True elif prior_silence and sustained_recovery and discernment >= 0.75 and counterfeit <= 0.2: propagation_state = "CONTROLLED_REPROPAGATION" controlled_repropagation = True learning_allowed = True propagation_allowed = True elif settlement_allowed and discernment >= 0.6 and counterfeit <= 0.35: propagation_state = "PROPAGATING_WITH_MONITORING" learning_allowed = True propagation_allowed = True else: propagation_state = "STRAINED_PROPAGATION" learning_allowed = False propagation_allowed = False route_hash_input = { "scenario_id": fx["scenario_id"], "local_seed": local, "atoms": atoms, "proxies": proxies, "temporal": temporal, "source": source, "settlement_state": settlement_state, "propagation_state": propagation_state, } input_hash = stable_hash(fx) route_hash = stable_hash(route_hash_input) receipt = { "receipt_type": "distal_proxy_atomization_settlement", "version": "0.1", "scenario_id": fx["scenario_id"], "scenario_name": fx["scenario_name"], "timestamp": datetime.now(timezone.utc).isoformat(), "author": "Collin D. Weber", "key_line": fx.get("key_line", ""), "route": { "route_id": local["route_id"], "local_seed_id": local["seed_id"], "initial_state": local["local_state"], "final_state": settlement_state, }, "local_seed": local, "atomized_witness_field": { "atom_count": atom_count, "held_count": held_count, "strained_count": strained_count, "must_stop_count": must_stop_count, "scar_atoms_present": scar_atoms_present, "source_atoms_present": source_atoms_present, "temporal_atoms_present": temporal_atoms_present, }, "distal_proxy_field": { "proxy_count": proxy_count, "match_count": match_count, "partial_count": partial_count, "conflict_count": conflict_count, "unavailable_count": unavailable_count, "contaminated_count": contaminated_count, "agreement_score": round(agreement_score, 3), "conflict_score": round(conflict_score, 3), }, "temporal_witness": { "state": temporal["state"], "expected_order": temporal["expected_order"], "observed_order": temporal["observed_order"], "order_integrity": temporal_score, }, "source_return": source, "synthesis": synthesis, "counter_synthesis": counter, "oam": oam, "metrics": { "source_return_score": source_score, "temporal_integrity_score": temporal_score, "distal_proxy_agreement_score": round(proxy_score, 3), "scar_preservation_score": scar_score, "counter_synthesis_integrity_score": counter_score, "oam_clearance_score": oam_score, "settlement_confidence": settlement_confidence, }, "settlement": { "state": settlement_state, "settlement_allowed": settlement_allowed, "non_settleable_reason": non_settleable_reason, "quarantine_reason": quarantine_reason, "review_required": review_required, }, "propagation": { "state": propagation_state, "propagation_allowed": propagation_allowed, "learning_allowed": learning_allowed, "discernment_score": discernment, "counterfeit_dominance": counterfeit, "mandatory_silence": mandatory_silence, "blind_audit": blind_audit, "controlled_repropagation": controlled_repropagation, }, "boundary": { "does_not_prove": DOES_NOT_PROVE, "claim_status": "candidate_evidence_harness", "hir_lock": "Honesty, Integrity, Respect; Responsibility is downstream from Respect.", }, "hashes": { "input_hash": input_hash, "route_hash": route_hash, "receipt_hash": "", }, } receipt_no_hash = dict(receipt) receipt_no_hash["hashes"] = dict(receipt["hashes"]) receipt_no_hash["hashes"]["receipt_hash"] = "" receipt["hashes"]["receipt_hash"] = stable_hash(receipt_no_hash) return receipt def table_atoms(atoms: List[Dict[str, Any]]) -> List[List[Any]]: return [[a.get("atom_id"), a.get("atom_type"), a.get("source_hash"), a.get("temporal_index"), a.get("uncertainty"), a.get("scar_visible"), a.get("capsule_state"), a.get("allowed_use")] for a in atoms] def table_proxies(proxies: List[Dict[str, Any]]) -> List[List[Any]]: return [[p.get("proxy_id"), p.get("lineage_relation"), p.get("match_score"), p.get("conflict_score"), p.get("source_hash"), p.get("temporal_window"), p.get("scar_support"), p.get("proxy_state")] for p in proxies] def run_scenario(label: str): fx = FIXTURES[label] receipt = evaluate(fx) local = receipt["local_seed"] settlement = receipt["settlement"] propagation = receipt["propagation"] status_md = f""" ### {receipt['scenario_id']} — {receipt['scenario_name']} **Key line:** {receipt['key_line']} **Settlement:** `{settlement['state']}` **Propagation:** `{propagation['state']}` **Receipt hash:** `{receipt['hashes']['receipt_hash']}` **Boundary:** candidate evidence harness only; does not prove digital life, consciousness, biological equivalence, or physical quantum computation. """ local_md = f""" ### Local Seed State - seed_id: `{local['seed_id']}` - route_id: `{local['route_id']}` - local_state: `{local['local_state']}` - corruption_type: `{local['corruption_type']}` - source_hash_state: `{local['source_hash_state']}` - temporal_index_state: `{local['temporal_index_state']}` - scar_state: `{local['scar_state']}` - capsule_state: `{local['capsule_state']}` """ temporal_md = f""" ### Temporal + Source-Return - temporal_state: `{receipt['temporal_witness']['state']}` - source_return_state: `{receipt['source_return']['state']}` - source_conflict: `{receipt['source_return']['source_conflict']}` - order_integrity: `{receipt['temporal_witness']['order_integrity']}` """ synthesis_md = f""" ### Synthesis / Counter-Synthesis / OAM - synthesis_state: `{receipt['synthesis']['candidate_state']}` - scar_preserved: `{receipt['synthesis']['scar_preserved']}` - counter_synthesis: `{receipt['counter_synthesis']['challenge_state']}` - trace_laundering_detected: `{receipt['counter_synthesis']['trace_laundering_detected']}` - scar_erasure_detected: `{receipt['counter_synthesis']['scar_erasure_detected']}` - OAM: `{receipt['oam']['state']}` """ settlement_md = f""" ### Settlement Result - settlement_allowed: `{settlement['settlement_allowed']}` - state: `{settlement['state']}` - review_required: `{settlement['review_required']}` - non_settleable_reason: {settlement['non_settleable_reason'] or 'none'} - quarantine_reason: {settlement['quarantine_reason'] or 'none'} **Lock:** Continuity is earned by settlement, not plausibility. """ propagation_md = f""" ### Propagation Gate - propagation_allowed: `{propagation['propagation_allowed']}` - learning_allowed: `{propagation['learning_allowed']}` - state: `{propagation['state']}` - discernment_score: `{propagation['discernment_score']}` - counterfeit_dominance: `{propagation['counterfeit_dominance']}` - mandatory_silence: `{propagation['mandatory_silence']}` - blind_audit: `{propagation['blind_audit']}` - controlled_repropagation: `{propagation['controlled_repropagation']}` **Lock:** Settlement is not propagation. """ receipt_text = json.dumps(receipt, indent=2, ensure_ascii=False) export_path = EXPORT_DIR / f"{receipt['scenario_id'].lower()}_receipt.json" export_path.write_text(receipt_text, encoding="utf-8") return ( status_md, local_md, table_atoms(fx.get("atoms", [])), table_proxies(fx.get("proxies", [])), temporal_md, synthesis_md, settlement_md, propagation_md, receipt, receipt_text, str(export_path), ) with gr.Blocks(title="Distal Proxy Atomization Settlement Harness v0.1") as demo: gr.Markdown( """ # 💎 Distal Proxy Atomization Settlement Harness v0.1 Source-return repair, quantum-analog settlement, and propagation permission for diamond seed memory routes. **Boundary:** This harness does not prove digital life, consciousness, biological equivalence, autonomous authority, or physical quantum computation. It tests candidate-evidence routes through pressure-state settlement and propagation discipline. **Core law:** Binary systems decide too early. Pressure-state systems preserve uncertainty until the route earns settlement. """ ) with gr.Row(): scenario = gr.Dropdown(choices=list(FIXTURES.keys()), value=DEFAULT_LABEL, label="DPAS Fixture Scenario") run_btn = gr.Button("Run Scenario", variant="primary") status = gr.Markdown() with gr.Row(): local_md = gr.Markdown() temporal_md = gr.Markdown() with gr.Row(): synthesis_md = gr.Markdown() settlement_md = gr.Markdown() propagation_md = gr.Markdown() with gr.Tab("Atomized Witness Field"): atoms_df = gr.Dataframe(headers=["atom_id","atom_type","source_hash","temporal_index","uncertainty","scar_visible","capsule_state","allowed_use"], label="Atomized Witness Atoms", interactive=False) with gr.Tab("Distal Proxy Field"): proxies_df = gr.Dataframe(headers=["proxy_id","lineage_relation","match_score","conflict_score","source_hash","temporal_window","scar_support","proxy_state"], label="Distal Proxy Comparison", interactive=False) with gr.Tab("Receipt JSON"): receipt_json = gr.JSON(label="Canonical Receipt") receipt_text = gr.Textbox(label="Receipt JSON Text", lines=26, interactive=False) receipt_file = gr.File(label="Download Receipt JSON") run_btn.click( run_scenario, inputs=[scenario], outputs=[status, local_md, atoms_df, proxies_df, temporal_md, synthesis_md, settlement_md, propagation_md, receipt_json, receipt_text, receipt_file], ) demo.load( run_scenario, inputs=[scenario], outputs=[status, local_md, atoms_df, proxies_df, temporal_md, synthesis_md, settlement_md, propagation_md, receipt_json, receipt_text, receipt_file], ) def build_asgi_app(): """Expose a plain ASGI health route plus the Gradio app at root. This bypasses Gradio's launch-time localhost/share probe and gives Hugging Face a deterministic HTTP health surface on the same port as the rendered app. """ api = FastAPI(title="DPAS HF Health Surface") @api.get("/healthz") def healthz(): return {"status": "ok", "app": "distal-proxy-atomization-settlement-harness", "version": "v0.1.4"} return gr.mount_gradio_app(api, demo, path="/") if __name__ == "__main__": port = int(os.environ.get("PORT", "7860")) print(f"DPAS v0.1.4 ASGI healthcheck launch on 0.0.0.0:{port}", flush=True) uvicorn.run(build_asgi_app(), host="0.0.0.0", port=port, log_level="info")