| 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") |
|
|
