"""Gradio interface for Iris."""
from __future__ import annotations
from dataclasses import dataclass, field
from html import escape
import json
from typing import Any, Iterable
from iris.engine import (
DirectionPressureResult,
FinalBrief,
IrisEngine,
PressureResult,
)
from iris.errors import IrisError
RINGS = 4
RING_NAMES = {
1: "Reality Contact",
2: "Real Actor",
3: "Existing Alternative",
4: "Problem Truth",
}
DIRECTION_NAMES = ("Constraints", "Limitations", "Capabilities", "Reality Contact")
# Cap how many prior pressure lines we replay so a small model can still attend
# to the freshest rings instead of drowning in a long trail.
MAX_TRAIL_LINES = 10
@dataclass(frozen=True)
class RingView:
depth: int
pressure: str
why_it_bites: str
alternative: str | None = None
@dataclass(frozen=True)
class CenterView:
actor: str
situation: str
assumption_to_test: str
next_step: str
@dataclass(frozen=True)
class SpiralView:
idea: str
rings: list[RingView]
center: CenterView | None = None
status: str = "ready"
message: str = ""
pending_depth: int | None = None
center_pending: bool = False
selected_depth: int | None = None
iteration: str | None = None
frame_title: str = "Idea frame 01"
@dataclass
class SpatialSession:
idea: str = ""
pressures: list[PressureResult] = field(default_factory=list)
center: CenterView | None = None
status: str = "ready"
message: str = ""
selected_depth: int | None = None
iteration: str | None = None
def create_app():
import gradio as gr
theme = gr.themes.Default(
font=("Georgia", "Times New Roman", "serif"),
font_mono=("ui-monospace", "SFMono-Regular", "Menlo", "monospace"),
)
with gr.Blocks(
title="Iris",
theme=theme,
css=APP_CSS,
js=APP_JS,
analytics_enabled=False,
fill_width=True,
) as demo:
gr.HTML(
render_interactive_canvas_html(),
elem_id="iris-stage",
container=False,
padding=False,
)
engine_request = gr.Textbox(visible=False, elem_id="iris-engine-request")
engine_response = gr.Textbox(visible=False, elem_id="iris-engine-response")
engine_trigger = gr.Button(
"Run Iris engine",
visible=False,
elem_id="iris-engine-trigger",
)
engine_trigger.click(
handle_canvas_request,
inputs=engine_request,
outputs=engine_response,
api_name="iris_canvas_engine",
show_progress="hidden",
)
return demo
def handle_canvas_request(request_json: str) -> str:
return run_canvas_engine(request_json)
def run_canvas_engine(
request_json: str,
*,
engine: IrisEngine | None = None,
) -> str:
try:
request = json.loads(request_json)
if not isinstance(request, dict):
raise ValueError("Engine request must be a JSON object.")
frame_id = str(request.get("frame_id", ""))
idea = _required_text(request, "idea")
mode = str(request.get("mode", "pressures"))
iterations = _idea_iterations(request.get("iterations", []))
prior_cards = _pressure_cards(request.get("prior_cards", []))
constraints = pressure_cards_to_constraints(prior_cards)
conversation = build_frame_conversation(iterations, prior_cards)
iris = engine or IrisEngine()
if mode == "finalize":
return _finalize_payload(
iris,
frame_id=frame_id,
current_idea=idea,
iterations=iterations,
prior_cards=prior_cards,
constraints=constraints,
conversation=conversation,
)
depth = _required_depth(request)
# The trail is replayed as real chat turns, so keep it out of the blob.
frame_idea = build_frame_context(
current_idea=idea,
iterations=iterations,
prior_cards=prior_cards,
include_trail=False,
)
total = max(depth + 1, RINGS)
results = iris.pressure_directions(
frame_idea,
constraints,
depth,
total,
allow_soft_failures=True,
conversation=conversation,
)
payload: dict[str, Any] = {
"ok": True,
"kind": "pressures",
"frame_id": frame_id,
"depth": depth,
"cards": [
{
"direction": result.direction,
"pressure": result.pressure,
"why_it_bites": result.why_it_bites,
}
for result in results
],
}
return json.dumps(payload)
except (IrisError, ValueError, TypeError, json.JSONDecodeError) as exc:
return json.dumps({"ok": False, "message": str(exc)})
def _finalize_payload(
iris: IrisEngine,
*,
frame_id: str,
current_idea: str,
iterations: list[dict[str, Any]],
prior_cards: list[dict[str, Any]],
constraints: list[str],
conversation: list[dict[str, str]],
) -> str:
frame_idea = build_frame_context(
current_idea=current_idea,
iterations=iterations,
prior_cards=prior_cards,
)
brief: FinalBrief = iris.finalize(
frame_idea, constraints, conversation=conversation
)
journey = [
{"version": item.get("version"), "idea": item.get("idea")}
for item in iterations
]
pressures = [
{
"depth": card.get("depth"),
"direction": card.get("direction"),
"pressure": card.get("pressure"),
"why_it_bites": card.get("why_it_bites"),
}
for card in prior_cards
if str(card.get("pressure", "")).strip()
]
center = brief.center
payload: dict[str, Any] = {
"ok": True,
"kind": "final",
"frame_id": frame_id,
"refined_idea": brief.refined_idea,
"journey": journey,
"pressures": pressures,
"next_step": center.next_step if center else "",
"actor": center.actor if center else "",
"situation": center.situation if center else "",
"assumption_to_test": center.assumption_to_test if center else "",
}
return json.dumps(payload)
def build_frame_conversation(
iterations: list[dict[str, Any]],
prior_cards: list[dict[str, Any]],
) -> list[dict[str, str]]:
"""Replay the frame as a real chat so the model sees its own prior pressures.
Each past idea iteration becomes a user turn, followed by an assistant turn
listing the pressures Iris already raised at that depth. This makes the model
treat the thread as one continuing, personalized conversation instead of a
cold one-shot prompt.
"""
sets: dict[int, list[dict[str, Any]]] = {}
for card in prior_cards:
try:
depth = int(card.get("depth", 0))
except (TypeError, ValueError):
depth = 0
sets.setdefault(depth, []).append(card)
messages: list[dict[str, str]] = []
for item in iterations:
idea = str(item.get("idea", "")).strip()
if not idea:
continue
try:
version = int(item.get("version", 0))
except (TypeError, ValueError):
version = 0
messages.append({"role": "user", "content": f"My idea (v{version}): {idea}"})
lines = []
for card in sets.get(version, []):
pressure = str(card.get("pressure", "")).strip()
if not pressure:
continue
direction = str(card.get("direction", "")).strip()
lines.append(f"{direction}: {pressure}" if direction else pressure)
if lines:
messages.append(
{
"role": "assistant",
"content": "Pressure I already raised:\n" + "\n".join(lines),
}
)
return messages
def build_frame_context(
*,
current_idea: str,
iterations: list[dict[str, Any]],
prior_cards: list[dict[str, Any]],
include_trail: bool = True,
) -> str:
history: list[tuple[int, str]] = []
for index, item in enumerate(iterations, start=1):
idea = str(item.get("idea", "")).strip()
if not idea:
continue
try:
version = int(item.get("version", index))
except (TypeError, ValueError):
version = index
history.append((version, idea))
if not history:
history.append((1, current_idea))
original_idea = history[0][1]
latest_version = history[-1][0]
# Lead with the root idea so a small model anchors on it, then the live
# iteration it must pressure now. Headings are load-bearing: the engine
# grounding logic parses "Original idea:", "Current iteration:", and
# "Iteration history:" sections by exact heading.
lines = [
"Frame continuity:",
(
"This is ONE continuous idea frame. Keep the original idea and every "
"later refinement in view. Pressure the current iteration as the "
"newest, sharpest version of the SAME original idea - build on it, "
"never abandon the original idea, and never repeat an earlier "
"pressure."
),
"",
"Original idea:",
original_idea,
"",
"Current iteration:",
current_idea,
"",
"Iteration history:",
]
for version, idea in history:
marker = " (current)" if version == latest_version else ""
lines.append(f"- Idea v{version}: {idea}{marker}")
# Cap the trail so the freshest pressure stays salient for a small model;
# keep the most recent entries (deepest rings) which matter most.
pressure_trail = _pressure_trail_lines(prior_cards) if include_trail else []
if pressure_trail:
trimmed = pressure_trail[-MAX_TRAIL_LINES:]
if len(pressure_trail) > MAX_TRAIL_LINES:
trimmed = [
f"- (earlier pressure trimmed: {len(pressure_trail) - MAX_TRAIL_LINES} "
"older entries)",
*trimmed,
]
lines.extend(
[
"",
"Prior AI pressure trail:",
*trimmed,
]
)
return "\n".join(lines)
def _pressure_trail_lines(cards: list[dict[str, Any]]) -> list[str]:
lines: list[str] = []
for card in cards:
pressure = str(card.get("pressure", "")).strip()
why = str(card.get("why_it_bites", "")).strip()
if not pressure and not why:
continue
try:
depth = int(card.get("depth", 0))
except (TypeError, ValueError):
depth = 0
depth_label = f"Depth {depth:02d}" if depth > 0 else "Prior depth"
direction = str(card.get("direction", "AI pressure")).strip() or "AI pressure"
if pressure and why:
lines.append(
f"- {depth_label} / {direction}: {pressure} Why it bites: {why}"
)
elif pressure:
lines.append(f"- {depth_label} / {direction}: {pressure}")
else:
lines.append(f"- {depth_label} / {direction} why_it_bites: {why}")
return lines
def pressure_cards_to_constraints(cards: list[dict[str, Any]]) -> list[str]:
constraints: list[str] = []
for card in cards:
pressure = str(card.get("pressure", "")).strip()
why = str(card.get("why_it_bites", "")).strip()
if not pressure or not why:
continue
alternative = str(card.get("alternative", "")).strip() or None
direction = str(card.get("direction", "")).strip()
if direction:
constraints.append(
DirectionPressureResult(
direction=direction,
pressure=pressure,
why_it_bites=why,
raw="",
).as_constraint()
)
else:
constraints.append(
PressureResult(
pressure=pressure,
why_it_bites=why,
raw="",
alternative=alternative,
).as_constraint()
)
return constraints
def _required_text(data: dict[str, Any], key: str) -> str:
value = data.get(key)
if not isinstance(value, str) or not value.strip():
raise ValueError(f"Expected non-empty string field: {key}")
return value.strip()
def _required_depth(data: dict[str, Any]) -> int:
value = data.get("depth")
if isinstance(value, bool):
raise ValueError("depth must be an integer")
try:
depth = int(value)
except (TypeError, ValueError) as exc:
raise ValueError("depth must be an integer") from exc
if depth < 1:
raise ValueError("depth must be positive")
return depth
def _pressure_cards(value: Any) -> list[dict[str, Any]]:
if value is None:
return []
if not isinstance(value, list):
raise ValueError("prior_cards must be a list")
cards: list[dict[str, Any]] = []
for item in value:
if isinstance(item, dict):
cards.append(item)
return cards
def _idea_iterations(value: Any) -> list[dict[str, Any]]:
if value is None:
return []
if not isinstance(value, list):
raise ValueError("iterations must be a list")
iterations: list[dict[str, Any]] = []
for index, item in enumerate(value, start=1):
if not isinstance(item, dict):
continue
idea = str(item.get("idea", "")).strip()
if not idea:
continue
try:
version = int(item.get("version", index))
except (TypeError, ValueError):
version = index
iterations.append({"version": version, "idea": idea})
return iterations
def render_interactive_canvas_html() -> str:
return """
No idea selectedIris
Ready.
Watch an idea sharpen under pressure.Click + New idea, drop one fuzzy idea, and let Iris apply pressure.Each idea becomes a scrolling thread. Switch ideas from the left rail.
"""
def stream_spiral(idea: str) -> Iterable[str]:
cleaned_idea = idea.strip()
if not cleaned_idea:
yield render_spiral_html(
SpiralView(
idea="",
rings=[],
status="error",
message="Enter one raw idea first.",
)
)
return
engine = IrisEngine()
pressures: list[PressureResult] = []
constraints: list[str] = []
yield render_spiral_html(
SpiralView(
idea=cleaned_idea,
rings=[],
status="running",
message=f"{ring_name(1)} is active.",
pending_depth=1,
)
)
try:
for depth in range(1, RINGS + 1):
result = engine.pressure(cleaned_idea, constraints, depth, RINGS)
pressures.append(result)
constraints.append(result.as_constraint())
next_depth = depth + 1 if depth < RINGS else None
yield render_spiral_html(
SpiralView(
idea=cleaned_idea,
rings=[
ring_to_view(index, item)
for index, item in enumerate(pressures, start=1)
],
status="running",
message=(
f"{ring_name(depth)} locked."
if next_depth is None
else f"{ring_name(next_depth)} is next."
),
pending_depth=next_depth,
)
)
yield render_spiral_html(
SpiralView(
idea=cleaned_idea,
rings=[
ring_to_view(index, item)
for index, item in enumerate(pressures, start=1)
],
status="running",
message="Center is forming.",
center_pending=True,
)
)
center = engine.distill(cleaned_idea, constraints)
except IrisError as exc:
yield render_spiral_html(
SpiralView(
idea=cleaned_idea,
rings=[
ring_to_view(index, item)
for index, item in enumerate(pressures, start=1)
],
status="error",
message=str(exc),
pending_depth=None,
center_pending=False,
)
)
return
yield render_spiral_html(
SpiralView(
idea=cleaned_idea,
rings=[
ring_to_view(index, item)
for index, item in enumerate(pressures, start=1)
],
center=CenterView(
actor=center.actor,
situation=center.situation,
assumption_to_test=center.assumption_to_test,
next_step=center.next_step,
),
status="complete",
message="Center reached.",
)
)
def ring_to_view(depth: int, result: PressureResult) -> RingView:
return RingView(
depth=depth,
pressure=result.pressure,
why_it_bites=result.why_it_bites,
alternative=result.alternative,
)
def ring_name(depth: int) -> str:
return RING_NAMES.get(depth, f"Ring {depth}")
def session_to_view(
session: SpatialSession,
*,
pending_depth: int | None = None,
center_pending: bool = False,
) -> SpiralView:
return SpiralView(
idea=session.idea,
rings=[
ring_to_view(index, item)
for index, item in enumerate(session.pressures, start=1)
],
center=session.center,
status=session.status,
message=session.message,
pending_depth=pending_depth,
center_pending=center_pending,
selected_depth=session.selected_depth,
iteration=session.iteration,
frame_title="Active pressure stack",
)
def render_spiral_html(view: SpiralView) -> str:
depth = len(view.rings)
status_text = escape(view.message or default_message(view))
status_class = safe_class_token(view.status)
depth_text = "Depth 00" if depth == 0 else f"Depth {depth:02d}"
return f"""
"""
def render_idea_card(view: SpiralView) -> str:
idea = view.idea or "Untitled idea"
return f"""
Idea v1User card
{escape(idea)}
"""
def render_ai_cards(view: SpiralView) -> str:
cards = [
render_ai_card(ring, selected=ring.depth == view.selected_depth)
for ring in view.rings
]
if view.pending_depth is not None:
cards.append(render_pending_ai_card(view.pending_depth))
if not cards:
cards.append(
"""
"""
def render_iteration_card(view: SpiralView) -> str:
iteration = view.iteration
if iteration is None:
iteration = "Next version of the idea lands here after the pressure cards."
return f"""
` : "";
const buttonDisabled = disabled ? "disabled" : "";
const placeholder =
entry.version === 1
? "e.g. an app that helps people remember names"
: "Sharpen it — rewrite the idea after the pressure…";
const body = entry.locked
? `