"
"
"
f""
f"{left_page}"
f"{right_page}"
"
"
"{_escape(input_preview)}
" "Output" f"{_escape(output_preview)}
" "" "" ) def _thread_modal(name: str, role: str, model: str, turns: list[dict[str, str]]) -> str: body = ( "".join( "{_escape(verdict.decree)}
" "{_escape(packet.title)}
" f"{_escape(packet.context or packet.setting)}
" f""
""
f"{_escape(subtitle)}
{_escape(item.excerpt)}
" f"Direction: {_escape(item.supports)}
" f"{_escape(item.note)}
The jurors are seated and silent.
Persona: {_escape(vote.persona)}
" f"{_escape(vote.vote.replace('_', ' '))}
" f"{_escape(vote.reason)}
" f"Evidence: {_escape(', '.join(vote.evidence_ids))}
{_escape(json.dumps(compact, indent=2))}"
def _clean_custom_items(values: list[str]) -> list[str]:
return [" ".join(value.split()) for value in values if " ".join(value.split())]
def _custom_case_from_payload(payload: str) -> CasePacket:
try:
data = json.loads(payload or "{}")
except json.JSONDecodeError as exc:
raise ValueError("Custom case details could not be read from the docket book.") from exc
context = " ".join(str(data.get("context", "")).split())
claimant_items = _clean_custom_items([str(value) for value in data.get("claimant_evidence", [])])
respondent_items = _clean_custom_items([str(value) for value in data.get("respondent_evidence", [])])
if not context:
raise ValueError("Custom requires a trial details paragraph.")
if not claimant_items or not respondent_items:
raise ValueError("Custom requires at least one evidence item for each side.")
evidence = [
EvidenceItem(
id=f"CUS-F{index}",
title=f"Claimant Evidence {index}",
source="Custom docket entry",
excerpt=item,
supports="claimant",
reliability=0.65,
note=item,
)
for index, item in enumerate(claimant_items[:3], start=1)
]
evidence.extend(
EvidenceItem(
id=f"CUS-A{index}",
title=f"Respondent Evidence {index}",
source="Custom docket entry",
excerpt=item,
supports="respondent",
reliability=0.65,
note=item,
)
for index, item in enumerate(respondent_items[:3], start=1)
)
return CasePacket(
id="custom",
title="Custom Trial",
subtitle="A custom docket assembled in the opening book.",
claimant="Claimant",
respondent="Respondent",
charge="Whether the custom record supports the claimant or the respondent.",
setting="A custom courtroom packet entered by the user.",
context=context,
claimant_claim="The claimant says the custom context and supporting evidence justify a favorable finding.",
respondent_claim="The respondent says the custom context is incomplete, overread, or answered by contrary evidence.",
source_note="Custom user-entered case packet from the docket book.",
evidence=evidence,
)
def render_case_preview(case_label: str) -> str:
case_id = CASE_OPTIONS.get(case_label, "socrates")
return render_court(
[],
pretrial_case=get_case(case_id) if case_id != "custom" else None,
custom_mode=case_id == "custom",
)
def run_ui(
case_label: str,
search_query: str,
hypothetical: str,
custom_payload: str,
speed: str,
mind_layer: bool,
):
case_id = CASE_OPTIONS.get(case_label, "socrates")
try:
custom_case = _custom_case_from_payload(custom_payload) if case_id == "custom" else None
except ValueError as exc:
yield (
render_court([], pretrial_case=None, custom_mode=True),
render_evidence([]),
render_jurors([]),
render_mind([], mind_layer),
str(exc),
)
return
request = TrialRequest(
case_id=case_id,
search_query=search_query or "",
hypothetical=hypothetical or "",
custom_case=custom_case,
speed=speed or "swift",
mind_layer=bool(mind_layer),
)
events: list[TrialEvent] = []
produced_events = _start_event_producer(request)
pending_speaker = _expected_next_speaker(events)
yield (
render_court(events, started=True, pending_speaker=pending_speaker),
render_evidence(events),
render_jurors(events),
render_mind(events, mind_layer),
_pending_status(pending_speaker),
)
try:
while True:
item = produced_events.get()
if item is _EVENT_STREAM_DONE:
break
if isinstance(item, Exception):
raise item
event = item
events.append(event)
yield (
render_court(events, started=True),
render_evidence(events),
render_jurors(events),
render_mind(events, mind_layer),
_event_status(event, len(events)),
)
duration = _reading_duration(_event_dialogue_text(event))
if duration > 0:
time.sleep(duration)
pending_speaker = _expected_next_speaker(events)
if pending_speaker is not None and produced_events.empty():
yield (
render_court(events, started=True, pending_speaker=pending_speaker),
render_evidence(events),
render_jurors(events),
render_mind(events, mind_layer),
_pending_status(pending_speaker),
)
except Exception as exc:
yield (
render_court(events, started=True),
render_evidence(events),
render_jurors(events),
render_mind(events, mind_layer),
f"Model response required. Trial stopped: {exc}",
)
return
yield (
render_court(events, started=True, show_verdict_popup=True),
render_evidence(events),
render_jurors(events),
render_mind(events, mind_layer),
"Verdict sealed.",
)
def build_app() -> gr.Blocks:
with gr.Blocks(title="Judge-GPT") as demo:
with gr.Group(elem_classes=["docket-book-controls"]):
gr.HTML("