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