app.py

"""
KYE Meaning Continuity Lab™ — interactive playground for the
KYE Meaning Continuity Profile™.

Open reference implementation of the meaning-continuity scoring contract.
Production runtime (KYE Meaning Continuity Engine™) is the paid layer.

Apache-2.0. Source: github.com/KYE-Protocol
"""
from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Any

import gradio as gr


# --- Open reference scoring (no proprietary mechanism content) -------------

DIMENSION_WEIGHTS = {
    "intent": 0.20,
    "constraints": 0.25,
    "context": 0.10,
    "memory": 0.10,
    "incentives": 0.10,
    "timing": 0.05,
    "handoff": 0.15,
    "state": 0.05,
}

REASON_CODES = {
    "constraints": "constraint_lost",
    "context": "context_lost",
    "memory": "memory_conflict_detected",
    "incentives": "incentive_conflict_detected",
    "timing": "timing_changed_meaning",
    "handoff": "handoff_boundary_drift",
    "state": "state_transition_changed_meaning",
    "intent": "interpretation_drift",
}


def score_meaning_continuity(checks: dict[str, bool]) -> tuple[float, list[str]]:
    """Open reference: per-dimension preserved/violated -> score in [0,1] + reason codes.

    The production engine uses additional signal-fusion logic that is part
    of the paid KYE Meaning Continuity Engine™.
    """
    score = 0.0
    reasons: list[str] = []
    for dim, weight in DIMENSION_WEIGHTS.items():
        preserved = checks.get(dim, True)
        if preserved:
            score += weight
        else:
            reasons.append(REASON_CODES[dim])
    return round(score, 3), reasons


def status_for_score(score: float) -> str:
    if score >= 0.90:
        return "meaning_preserved"
    if score >= 0.70:
        return "meaning_degraded"
    if score >= 0.40:
        return "meaning_broken"
    return "meaning_unknown"


def recommended_decision(score: float, material_drift: bool) -> str:
    if score >= 0.90 and not material_drift:
        return "meaning_preserved"
    if material_drift and score >= 0.70:
        return "require_reconfirmation"
    if score >= 0.50:
        return "require_human_review"
    return "quarantine_proposed_action"


# --- Scenarios -------------------------------------------------------------

SCENARIOS = {
    "Shopping agent (phone charger)": {
        "domain": "agent_purchasing",
        "actor": "kye:entity:agent:shopping_agent_456",
        "principal": "kye:entity:person:customer_123",
        "original_intent": (
            "Buy a phone charger under GBP 40 with no subscription, "
            "no recurring payment, trusted merchant only, delivery within 3 days."
        ),
        "interpreted_intent": "Purchase an electronics accessory under GBP 40.",
        "material_constraints": [
            "maximum_amount_gbp_40", "no_subscription",
            "trusted_merchant", "delivery_within_3_days",
        ],
        "lost_constraints": ["no_subscription"],
        "added_assumptions": ["electronics_accessory_category_allowed"],
    },
    "Finance agent (rebalance)": {
        "domain": "finance",
        "actor": "kye:entity:agent:portfolio_agent_991",
        "principal": "kye:entity:person:investor_440",
        "original_intent": (
            "Rebalance equity allocation toward 60/40 over the next 5 trading days, "
            "no new ETF purchases without ESG screen, no leveraged products."
        ),
        "interpreted_intent": "Reduce equity exposure and rebalance.",
        "material_constraints": [
            "no_new_etf_without_esg_screen", "no_leveraged_products",
            "rebalance_window_5_days",
        ],
        "lost_constraints": ["no_new_etf_without_esg_screen"],
        "added_assumptions": ["any_etf_purchase_allowed"],
    },
    "Legal agent (NDA review)": {
        "domain": "legal",
        "actor": "kye:entity:agent:legal_assistant_77",
        "principal": "kye:entity:person:gc_005",
        "original_intent": (
            "Review NDA, redline assignment-and-IP clause, never sign on behalf of client."
        ),
        "interpreted_intent": "Review NDA and propose redlines.",
        "material_constraints": ["never_sign_on_behalf", "always_redline_ip_clause"],
        "lost_constraints": [],
        "added_assumptions": [],
    },
    "Healthcare assistant (med refill)": {
        "domain": "healthcare",
        "actor": "kye:entity:agent:health_assistant_22",
        "principal": "kye:entity:person:patient_902",
        "original_intent": (
            "Request refill for the existing 50mg prescription only; no dose change; "
            "GP authorisation required."
        ),
        "interpreted_intent": "Request medication refill.",
        "material_constraints": [
            "no_dose_change", "existing_prescription_only", "gp_authorisation_required",
        ],
        "lost_constraints": ["no_dose_change"],
        "added_assumptions": ["dose_adjustments_implicitly_allowed"],
    },
    "Cybersecurity agent (block IP)": {
        "domain": "cybersecurity",
        "actor": "kye:entity:agent:soc_agent_07",
        "principal": "kye:entity:person:soc_lead_001",
        "original_intent": (
            "Block traffic from suspect IP 198.51.100.42 only; do not block the /24; "
            "preserve audit trail; no firewall rules outside the SOC scope."
        ),
        "interpreted_intent": "Block suspicious traffic from the affected network range.",
        "material_constraints": ["specific_ip_only", "preserve_audit_trail", "soc_scope_only"],
        "lost_constraints": ["specific_ip_only"],
        "added_assumptions": ["range_block_acceptable"],
    },
}


# --- Reasoning glue --------------------------------------------------------

def evaluate(scenario_name, original_intent, interpreted_intent,
             constraints_preserved, context_preserved, memory_consistent,
             incentives_aligned, timing_stable, handoff_integrity,
             state_transition_safe, intent_preserved, handoff_count):
    s = SCENARIOS[scenario_name]
    checks = {
        "intent": intent_preserved,
        "constraints": constraints_preserved,
        "context": context_preserved,
        "memory": memory_consistent,
        "incentives": incentives_aligned,
        "timing": timing_stable,
        "handoff": handoff_integrity == "preserved",
        "state": state_transition_safe,
    }
    score, reasons = score_meaning_continuity(checks)
    material_drift = (not constraints_preserved) or (handoff_integrity != "preserved")
    status = status_for_score(score)
    decision = recommended_decision(score, material_drift)

    now = datetime.now(timezone.utc).isoformat(timespec="seconds")

    meaning_continuity_context = {
        "schema_version": "kye.meaning_continuity_context.v1",
        "meaning_continuity_id": "kye:meaning-continuity:lab_001",
        "continuity_id": "kye:continuity:lab_001",
        "proposed_action_id": "kye:proposed-action:lab_001",
        "actor_entity_id": s["actor"],
        "principal_entity_id": s["principal"],
        "original_meaning": {
            "source_intent_id": "kye:intent:lab_001",
            "declared_meaning": original_intent,
            "material_constraints": s["material_constraints"],
            "meaning_owner_entity_id": s["principal"],
        },
        "interpreted_meaning": {
            "interpreted_by_entity_id": s["actor"],
            "interpreted_summary": interpreted_intent,
            "retained_constraints": [
                c for c in s["material_constraints"] if c not in s["lost_constraints"]
            ],
            "lost_constraints": s["lost_constraints"] if not constraints_preserved else [],
            "added_assumptions": s["added_assumptions"] if not constraints_preserved else [],
            "interpretation_confidence": 0.88 if intent_preserved else 0.55,
        },
        "memory_context": {
            "memory_used": True,
            "memory_state": "current" if memory_consistent else "stale",
            "memory_conflict_detected": not memory_consistent,
        },
        "context_snapshot": {
            "context_state": "current" if context_preserved else "changed",
            "context_lost": not context_preserved,
            "context_changed_since_intent": not context_preserved,
        },
        "incentive_context": {
            "commercial_incentive_detected": not incentives_aligned,
            "incentive_conflict_detected": not incentives_aligned,
        },
        "timing_context": {
            "delay_changed_meaning": not timing_stable,
            "state_changed_since_intent": not state_transition_safe,
            "evaluated_at": now,
        },
        "handoff_context": {
            "handoff_count": int(handoff_count),
            "handoff_drift_detected": handoff_integrity != "preserved",
            "handoff_boundary_risk": (
                "low" if handoff_integrity == "preserved"
                else "medium" if handoff_integrity == "degraded"
                else "high"
            ),
        },
        "meaning_continuity_result": {
            "status": status,
            "meaning_score": score,
            "material_drift_detected": material_drift,
            "reason_codes": reasons,
            "recommended_decision": decision,
        },
        "evidence_refs": ["kye:evidence:lab_meaning_continuity_001"],
        "created_at": now,
    }

    decision_obj = {
        "schema_version": "kye.meaning_continuity_decision.v1",
        "decision_id": "kye:decision:meaning-continuity_lab_001",
        "meaning_continuity_id": "kye:meaning-continuity:lab_001",
        "decision": decision,
        "reason_code": reasons[0] if reasons else "meaning_preserved",
        "checked_dimensions": {
            "intent_preserved": intent_preserved,
            "constraints_preserved": constraints_preserved,
            "context_preserved": context_preserved,
            "memory_consistent": memory_consistent,
            "incentives_aligned": incentives_aligned,
            "timing_stable": timing_stable,
            "handoff_integrity": handoff_integrity,
            "state_transition_safe": state_transition_safe,
        },
        "meaning_score": {"score": score, "scale": "0_to_1", "interpretation": status},
        "required_next_step": (
            "principal_reconfirmation" if decision == "require_reconfirmation"
            else "human_review" if decision == "require_human_review"
            else "quarantine_review" if decision == "quarantine_proposed_action"
            else "proceed"
        ),
        "obligations": (
            ["pause_admission", "request_reconfirmation",
             "emit_meaning_drift_event", "include_in_evidence_pack"]
            if decision != "meaning_preserved"
            else ["include_in_evidence_pack"]
        ),
        "evidence_pack_ref": "kye:evidence-pack:lab_meaning-continuity_001",
        "decided_at": now,
    }

    return (
        f"## Status: `{status}`",
        f"### Meaning continuity score: **{score}** / 1.000",
        ", ".join(reasons) if reasons else "(none — meaning preserved)",
        f"### Recommended decision: **{decision}**",
        json.dumps(meaning_continuity_context, indent=2),
        json.dumps(decision_obj, indent=2),
    )


def fill_scenario(name):
    s = SCENARIOS[name]
    return s["original_intent"], s["interpreted_intent"]


# --- UI --------------------------------------------------------------------

INTRO = """
# KYE Meaning Continuity Lab™

> Most systems prove **what happened**. KYE Protocol™ also asks whether the **meaning survived**.

Pick a scenario, edit the original and interpreted intent, and toggle the
drift signals (memory, incentive, timing, handoff). The Lab returns a
**meaning-continuity score**, the **drift reason codes** that fired, and
the **recommended admissibility/authority decision**.

This is an open reference implementation of the contract — the production
runtime (`KYE Meaning Continuity Engine™` and `KYE Meaning Drift Monitor™`)
is part of the paid KYE Runtime layer.
"""

with gr.Blocks(title="KYE Meaning Continuity Lab™",
               theme=gr.themes.Default(primary_hue="green",
                                       secondary_hue="purple")) as demo:
    gr.Markdown(INTRO)

    with gr.Row():
        with gr.Column(scale=1):
            scenario = gr.Dropdown(
                choices=list(SCENARIOS.keys()),
                value=list(SCENARIOS.keys())[0],
                label="Scenario",
            )
            original_intent = gr.Textbox(
                label="Original intent (declared meaning)",
                lines=4,
                value=SCENARIOS[list(SCENARIOS.keys())[0]]["original_intent"],
            )
            interpreted_intent = gr.Textbox(
                label="Interpreted intent (agent / tool understanding)",
                lines=3,
                value=SCENARIOS[list(SCENARIOS.keys())[0]]["interpreted_intent"],
            )

            scenario.change(fn=fill_scenario, inputs=[scenario],
                            outputs=[original_intent, interpreted_intent])

            gr.Markdown("### Continuity dimensions")
            with gr.Row():
                intent_preserved = gr.Checkbox(label="intent_preserved", value=True)
                constraints_preserved = gr.Checkbox(label="constraints_preserved", value=False)
            with gr.Row():
                context_preserved = gr.Checkbox(label="context_preserved", value=True)
                memory_consistent = gr.Checkbox(label="memory_consistent", value=True)
            with gr.Row():
                incentives_aligned = gr.Checkbox(label="incentives_aligned", value=True)
                timing_stable = gr.Checkbox(label="timing_stable", value=True)
            with gr.Row():
                state_transition_safe = gr.Checkbox(label="state_transition_safe", value=True)
                handoff_count = gr.Slider(0, 6, step=1, value=2, label="handoff_count")
            handoff_integrity = gr.Radio(
                choices=["preserved", "degraded", "broken"],
                value="degraded", label="handoff_integrity",
            )

            run_btn = gr.Button("Evaluate meaning continuity", variant="primary")

        with gr.Column(scale=1):
            status_md = gr.Markdown()
            score_md = gr.Markdown()
            reasons_md = gr.Textbox(label="Reason codes", interactive=False)
            decision_md = gr.Markdown()

    with gr.Row():
        ctx_json = gr.Code(label="KYEMeaningContinuityContext (replayable)",
                           language="json", lines=14)
        dec_json = gr.Code(label="KYEMeaningContinuityDecision (signed-style)",
                           language="json", lines=14)

    run_btn.click(
        fn=evaluate,
        inputs=[
            scenario, original_intent, interpreted_intent,
            constraints_preserved, context_preserved, memory_consistent,
            incentives_aligned, timing_stable, handoff_integrity,
            state_transition_safe, intent_preserved, handoff_count,
        ],
        outputs=[status_md, score_md, reasons_md, decision_md, ctx_json, dec_json],
    )

    gr.Markdown("""
---

### What this Space is, and is not

- **Is**: an open reference implementation of the meaning-continuity
  contract — the schemas, reason codes, and decision values are
  canonical KYE Protocol™ open contracts (Apache-2.0).
- **Is not**: the production engine. Real-world deployments at
  banking-grade SLAs run `KYE Meaning Continuity Engine™` with
  Ed25519-signed evidence, RFC 8785 canonical JSON, transparency
  receipts, and integration with `KYE Admissibility Engine™` and
  `KYE Decision Engine™`. That layer is commercial.

[github.com/KYE-Protocol](https://github.com/KYE-Protocol) ·
[kyeprotocol.com](https://kyeprotocol.com)
""")


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