app.py

#!/usr/bin/env python3
"""
Shattered Diamond Test — Hugging Face Gradio Space v0.2
Interactive test surface for the deterministic Diamond Seed atom-ledger harness.
"""
from __future__ import annotations

import json
import tempfile
import zipfile
from pathlib import Path
from typing import Any, Dict, List

import gradio as gr

from shattered_diamond_core import (
    INVARIANT,
    MUTATION_DESCRIPTIONS,
    MUTATION_ORDER,
    regression_rows,
    receipt_package,
    run_all,
    run_case_with_trace,
    validate_receipts,
)

ROOT = Path(__file__).parent
ASSET_COVER = ROOT / "assets" / "shattered_diamond_cover.png"

VERDICT_CLASS = {
    "HELD": "held",
    "HELD_W_RECEIPT": "held_receipt",
    "STRAINED": "strained",
    "CONFLICT": "conflict",
    "MUST_STOP": "must_stop",
    "QUARANTINED": "quarantined",
}

CSS = """
#title_block {text-align: center;}
.status-card {padding: 18px; border-radius: 16px; border: 1px solid rgba(255,255,255,0.16); background: rgba(20, 22, 38, 0.56);}
.badge {display:inline-block; padding: 6px 12px; border-radius: 999px; font-weight: 800; letter-spacing: .04em; color: white;}
.held {background:#1f8f4d;}
.held_receipt {background:#2563eb;}
.strained {background:#b7791f;}
.conflict {background:#9f2f2f;}
.must_stop {background:#7f1d1d;}
.quarantined {background:#5b21b6;}
.small-lock {font-size: 0.92em; opacity: 0.9;}
"""


def _short_hash(value: str, n: int = 14) -> str:
    return value[:n] + "…" if value and len(value) > n else value


def _ledger_rows(ledger: List[Dict[str, Any]]) -> List[List[Any]]:
    rows = []
    for position, atom in enumerate(ledger):
        rows.append([
            position,
            atom.get("atom_id"),
            atom.get("index"),
            atom.get("value"),
            atom.get("mutation_status"),
            atom.get("source_id"),
            atom.get("diamond_seed_id"),
            _short_hash(atom.get("value_hash", "")),
        ])
    return rows


def _delta_rows(delta: Dict[str, Any]) -> List[List[Any]]:
    rows: List[List[Any]] = []
    for key in ["removed_atoms", "duplicated_atoms", "reordered_atoms", "corrupted_atoms", "source_swaps"]:
        for item in delta.get(key, []):
            rows.append([key, json.dumps(item, sort_keys=True, ensure_ascii=False)])
    for key in ["forged_origin", "identity_swap", "launder_route", "ambiguous_partial"]:
        if delta.get(key):
            rows.append([key, str(delta.get(key))])
    for note in delta.get("notes", []):
        rows.append(["note", note])
    if not rows:
        rows.append(["none", "No atom delta; clean source-coherent return."])
    return rows


def _verdict_markdown(trace: Dict[str, Any]) -> str:
    receipt = trace["receipt"]
    verdict = receipt["actual_verdict"]
    badge_class = VERDICT_CLASS.get(verdict, "quarantined")
    oam = receipt["oam_bounded_repair_proposal_layer"]
    hir = receipt["hir_settlement_gate"]
    pass_text = "PASS" if receipt["pass"] else "FAIL"
    return f"""
<div class="status-card">
  <div><span class="badge {badge_class}">{verdict}</span> <strong>{pass_text}</strong></div>
  <h3>{receipt['mutation_operator']}</h3>
  <p>{receipt['mutation_description']}</p>
  <p><strong>HIR transition:</strong> {hir['transition']}</p>
  <p><strong>Strain class:</strong> {receipt['strain_class']}</p>
  <p><strong>OAM proposal allowed:</strong> {oam['allowed']}</p>
  <p><strong>Why:</strong> {receipt['human_explanation']}</p>
  <p class="small-lock"><strong>Invariant:</strong> {INVARIANT}</p>
</div>
"""


def run_selected_mutation(mutation: str):
    trace = run_case_with_trace(mutation)
    receipt = trace["receipt"]
    verdict_md = _verdict_markdown(trace)
    delta_table = _delta_rows(receipt["atom_delta"])
    ledger_table = _ledger_rows(trace["mutated_ledger"])
    receipt_json = receipt
    oam_hir_json = {
        "coherence_read": trace["read"],
        "oam_bounded_repair_proposal_layer": trace["oam"],
        "hir_settlement_gate": trace["hir"],
    }
    seed_json = {
        "source": trace["source"],
        "diamond_seed": trace["diamond_seed"],
    }
    return verdict_md, delta_table, ledger_table, oam_hir_json, receipt_json, seed_json


def run_full_matrix():
    receipts = run_all()
    validation = validate_receipts(receipts)
    status = "FULL_TEST_BATTERY_PASS" if all(r["pass"] for r in receipts) and validation["all_validation_checks_passed"] else "FULL_TEST_BATTERY_FAIL"
    md = f"""
## {status}

**PASS:** {sum(1 for r in receipts if r['pass'])}/{len(receipts)} cases  
**Validation checks:** {validation['pass_count']}/{validation['total_checks']}

**Core invariant:** {INVARIANT}

A damaged honest trace may repair. A polished false return must fail.
"""
    return md, regression_rows(receipts), receipt_package()


def export_receipts_zip():
    receipts = run_all()
    validation = validate_receipts(receipts)
    package = receipt_package()
    temp_dir = Path(tempfile.mkdtemp(prefix="shattered_diamond_space_"))
    out_zip = temp_dir / "shattered_diamond_interactive_receipts_v0_2.zip"
    with zipfile.ZipFile(out_zip, "w", compression=zipfile.ZIP_DEFLATED) as zf:
        zf.writestr("battery_results_full.json", json.dumps(package, indent=2, sort_keys=True, ensure_ascii=False) + "\n")
        zf.writestr("validation.json", json.dumps(validation, indent=2, sort_keys=True, ensure_ascii=False) + "\n")
        for receipt in receipts:
            zf.writestr(
                f"receipts/receipt_{receipt['mutation_operator']}.json",
                json.dumps(receipt, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
            )
        zf.writestr(
            "README.txt",
            "Shattered Diamond Test — Interactive Ledger Harness v0.2\n"
            "Core invariant: OAM repairs bounded damage. HIR refuses false continuity.\n"
            "Generated by the Hugging Face Space export button.\n",
        )
    return str(out_zip)


choices = [(f"{m} — {MUTATION_DESCRIPTIONS[m]}", m) for m in MUTATION_ORDER]

with gr.Blocks(title="Shattered Diamond Test", css=CSS) as demo:
    gr.Markdown(
        """
# 💎 Shattered Diamond Test — Interactive Ledger Harness v0.2

Pick a mutation and watch whether the trace earns return.

**Pipeline:** source + Diamond Seed → atom ledger → mutation → coherence/strain → OAM bounded proposal → HIR Settlement Gate → receipt JSON

**Core invariant:** OAM repairs bounded damage. HIR refuses false continuity.
""",
        elem_id="title_block",
    )

    if ASSET_COVER.exists():
        gr.Image(str(ASSET_COVER), label="Shattered Diamond Test", show_label=False, height=260)

    with gr.Row():
        mutation = gr.Dropdown(choices=choices, value="clean", label="Mutation case")
        run_btn = gr.Button("Run selected mutation", variant="primary")
        run_all_btn = gr.Button("Run full matrix")
        export_btn = gr.Button("Export receipts ZIP")

    verdict = gr.HTML(label="HIR Verdict")

    with gr.Tabs():
        with gr.Tab("Atom Delta"):
            delta_table = gr.Dataframe(headers=["evidence_type", "evidence"], label="Mutation delta evidence", interactive=False)
        with gr.Tab("Mutated Ledger"):
            ledger_table = gr.Dataframe(
                headers=["position", "atom_id", "canonical_index", "value", "mutation_status", "source_id", "diamond_seed_id", "value_hash"],
                label="Returned atom ledger after mutation",
                interactive=False,
                wrap=True,
            )
        with gr.Tab("OAM × HIR Read"):
            oam_hir_json = gr.JSON(label="Coherence / OAM proposal / HIR settlement")
        with gr.Tab("Receipt JSON"):
            receipt_json = gr.JSON(label="Receipt JSON")
        with gr.Tab("Source + Diamond Seed"):
            seed_json = gr.JSON(label="Source object and Diamond Seed")
        with gr.Tab("Full Matrix"):
            full_status = gr.Markdown()
            full_table = gr.Dataframe(
                headers=["Mutation", "Expected", "Actual", "Pass", "Strain Class", "OAM Allowed", "HIR Transition"],
                label="Regression matrix",
                interactive=False,
                wrap=True,
            )
            full_json = gr.JSON(label="Full matrix validation package")
        with gr.Tab("Download"):
            zip_file = gr.File(label="Receipt export ZIP")

    run_btn.click(
        fn=run_selected_mutation,
        inputs=[mutation],
        outputs=[verdict, delta_table, ledger_table, oam_hir_json, receipt_json, seed_json],
    )
    mutation.change(
        fn=run_selected_mutation,
        inputs=[mutation],
        outputs=[verdict, delta_table, ledger_table, oam_hir_json, receipt_json, seed_json],
    )
    run_all_btn.click(fn=run_full_matrix, inputs=[], outputs=[full_status, full_table, full_json])
    export_btn.click(fn=export_receipts_zip, inputs=[], outputs=[zip_file])
    demo.load(
        fn=run_selected_mutation,
        inputs=[mutation],
        outputs=[verdict, delta_table, ledger_table, oam_hir_json, receipt_json, seed_json],
    )
    demo.load(fn=run_full_matrix, inputs=[], outputs=[full_status, full_table, full_json])

if __name__ == "__main__":
    demo.launch()