app.py

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