| """Executive Dynamics workspace — the analyst-facing "what changed during this call" |
| view: a call list, a per-turn behavioral Timeline (the hero), an Evidence panel, and |
| a synced Transcript. Reuses transcript.py's rendering and call_filters.py's filtering |
| rather than reimplementing them. |
| |
| Click interaction: a single shared `{"turn_index": int}` event, fired via |
| `gr.HTML(js_on_load=...)` + the browser-side `trigger()` helper Gradio 6 injects into |
| every HTML component, from BOTH the Timeline and the Transcript panel, into one |
| `on_turn_click` reducer — this is what makes Timeline <-> Transcript <-> Evidence |
| sync "for free": same event shape, one state update, three re-renders. |
| |
| IMPORTANT (bug fixed here): `js_on_load` executes exactly ONCE, when the component |
| first mounts — it does NOT re-run on later `value` updates (confirmed by reading the |
| compiled frontend, gradio/templates/frontend/assets/HTML-*.js: the mount effect is |
| gated by a "has-run" flag). Binding `click` listeners directly to `[data-turn-index]` |
| elements would therefore only ever bind to whatever was on screen at mount time (an |
| empty Timeline, since no call is selected yet) — clicks on turns rendered later would |
| silently do nothing. The fix is event delegation: bind ONE listener to `element` |
| itself (the persistent outer container that survives every content swap; only its |
| children get diffed/patched) and use `event.target.closest('[data-turn-index]')` to |
| resolve the actual turn at click time, however many times the inner HTML has changed |
| since mount. |
| """ |
|
|
| import html |
| import random |
|
|
| import gradio as gr |
| import pandas as pd |
|
|
| from call_filters import TABLE_HEADERS, filter_calls, filter_calls_df |
| from data_loader import load_call_detail, load_ticker_names |
| from product_language import ( |
| company_dropdown_choices, |
| largest_friction_label, |
| risk_driver_bullets, |
| risk_tier_color, |
| risk_tier_label, |
| ) |
| from theme import CLASS_COLORS, CLASS_NAMES, SIG_NEGATIVE, SIG_NEUTRAL, SIG_POSITIVE |
| from transcript import render_full_transcript |
|
|
| QA_SEGMENT_TYPES = {"analyst_question", "management_answer"} |
|
|
| |
| |
| |
| |
| _DEFAULT_COVERAGE = "OOF-covered only" |
|
|
| DIMENSION_ROWS = [ |
| ("confidence", "Confidence"), |
| ("stress", "Stress"), |
| ("specificity", "Specificity"), |
| ("alignment_drift", "Alignment"), |
| ] |
|
|
| _CLICK_BINDING_JS = ( |
| "element.addEventListener('click', (e) => {" |
| " const node = e.target.closest('[data-turn-index]');" |
| " if (node) { trigger('click', {turn_index: parseInt(node.dataset.turnIndex, 10)}); }" |
| "});" |
| ) |
|
|
|
|
| def _turn_html_wrapper(inner: str, js_on_load: str = _CLICK_BINDING_JS, elem_id: str | None = None, elem_classes=None): |
| return gr.HTML(value=inner, js_on_load=js_on_load, elem_id=elem_id, elem_classes=elem_classes) |
|
|
|
|
| def render_summary_strip(detail: dict | None) -> str: |
| """Behavioral read on the call — deliberately excludes Confidence, which already |
| has its own tile in the Verdict card above; repeating it here would just be the |
| same number shown twice a few inches apart.""" |
| if detail is None: |
| return '<div class="fl-summary-strip"><div class="fl-stat-block"><div class="fl-evidence-empty">Select a call to see its behavioral summary.</div></div></div>' |
|
|
| consistency = detail.get("behavioral_consistency") |
| consistency_str = f"{consistency:.0%}" if consistency is not None else "—" |
| tier = detail.get("risk_tier") |
| tier_color = risk_tier_color(tier) |
| friction_label = largest_friction_label(detail.get("largest_friction_dimension")) |
|
|
| return ( |
| '<div class="fl-summary-strip">' |
| f'<div class="fl-stat-block"><div class="fl-stat-label">Behavioral Consistency</div>' |
| f'<div class="fl-stat-value">{consistency_str}</div></div>' |
| f'<div class="fl-stat-block"><div class="fl-stat-label">Risk Tier</div>' |
| f'<div class="fl-stat-value" style="color:{tier_color};">{risk_tier_label(tier)}</div></div>' |
| f'<div class="fl-stat-block"><div class="fl-stat-label">Largest Friction</div>' |
| f'<div class="fl-stat-value">{friction_label}</div></div>' |
| "</div>" |
| ) |
|
|
|
|
| def render_timeline(turns: list[dict], selected_turn: int | None) -> str: |
| if not turns: |
| return '<div class="fl-timeline"><div class="fl-evidence-empty">Select a call to see its behavioral timeline.</div></div>' |
|
|
| n = len(turns) |
| qa_start_index = next( |
| (t["turn_index"] for t in turns if t["segment_type"] in QA_SEGMENT_TYPES), None |
| ) |
| divider_pct = (qa_start_index / n * 100) if qa_start_index is not None else None |
|
|
| rows_html = [] |
| for dim_key, dim_label in DIMENSION_ROWS: |
| bars = [] |
| for t in turns: |
| value = t.get(dim_key) |
| if value is None: |
| continue |
| pct = t["turn_index"] / max(n - 1, 1) * 100 |
| height_pct = max(value, 0.02) * 100 |
| selected_cls = " fl-timeline-selected" if t["turn_index"] == selected_turn else "" |
| bars.append( |
| f'<div class="fl-timeline-bar{selected_cls}" data-turn-index="{t["turn_index"]}" ' |
| f'style="left:{pct:.2f}%; height:{height_pct:.1f}%;" ' |
| f'title="{dim_label}: {value:.2f}"></div>' |
| ) |
| divider_html = ( |
| f'<div class="fl-timeline-divider" style="left:{divider_pct:.2f}%;"></div>' |
| if divider_pct is not None else "" |
| ) |
| rows_html.append( |
| f'<div class="fl-timeline-row">' |
| f'<div class="fl-timeline-label">{dim_label}</div>' |
| f'<div class="fl-timeline-strip">{"".join(bars)}{divider_html}</div>' |
| f"</div>" |
| ) |
|
|
| risk_dots = [] |
| for t in turns: |
| if not t.get("is_risk_event"): |
| continue |
| pct = t["turn_index"] / max(n - 1, 1) * 100 |
| selected_cls = " fl-timeline-selected" if t["turn_index"] == selected_turn else "" |
| risk_dots.append( |
| f'<div class="fl-risk-dot{selected_cls}" data-turn-index="{t["turn_index"]}" ' |
| f'style="left:{pct:.2f}%;" title="Risk event"></div>' |
| ) |
|
|
| section_labels = "" |
| if divider_pct is not None: |
| |
| |
| |
| |
| |
| |
| section_labels = f'<div class="fl-timeline-section-label" style="left:{divider_pct:.2f}%;">Q&A</div>' |
| if divider_pct >= 22: |
| section_labels = ( |
| '<div class="fl-timeline-section-label" style="left:0%;">Prepared Remarks</div>' |
| + section_labels |
| ) |
|
|
| return ( |
| '<div class="fl-timeline">' |
| f'<div style="position:relative;">{section_labels}</div>' |
| + "".join(rows_html) |
| + f'<div class="fl-timeline-risk-row">{"".join(risk_dots)}</div>' |
| + "</div>" |
| ) |
|
|
|
|
| def render_evidence(turn: dict | None) -> str: |
| if turn is None: |
| return '<div class="fl-evidence-panel"><div class="fl-evidence-empty">Click a point on the Timeline or a transcript turn to inspect it.</div></div>' |
|
|
| speaker = html.escape(turn.get("speaker_name") or turn["speaker_role"]) |
| quote = html.escape(turn["text"][:400]) + ("…" if len(turn["text"]) > 400 else "") |
| bullets = risk_driver_bullets(turn) |
| bullets_html = ( |
| '<ul class="fl-evidence-bullets fl-mono" style="color: var(--fl-accent);">' + "".join(f'<li class="fl-evidence-bullet" style="margin-bottom: 4px;">> {html.escape(b).upper()}</li>' for b in bullets) + "</ul>" |
| if bullets else '<div class="fl-evidence-empty">No elevated friction on this turn.</div>' |
| ) |
| is_risk = turn.get("is_risk_event", False) |
| badge_color = SIG_NEGATIVE if is_risk else SIG_POSITIVE |
| badge_text = "Dissonance Detected" if is_risk else "Aligned" |
| minutes, seconds = divmod(int(turn.get("start_time") or 0), 60) |
|
|
| return ( |
| '<div class="fl-evidence-panel">' |
| f'<div class="fl-evidence-meta">{minutes:02d}:{seconds:02d} · {speaker} ({turn["speaker_role"]}) ' |
| f'· <span style="color:{badge_color};">{badge_text}</span></div>' |
| f'<div class="fl-evidence-quote fl-serif-text" style="font-size: 18px;">“{quote}”</div>' |
| f"{bullets_html}" |
| "</div>" |
| ) |
|
|
|
|
| def render_transcript_synced(turns: list[dict], selected_turn: int | None) -> str: |
| return render_full_transcript(turns, selected_turn=selected_turn) |
|
|
|
|
| def render_verdict_card(detail: dict | None, call_id: str | None) -> str: |
| """The punchline: what the sentiment model predicted for this call vs. the |
| reference label, surfaced as its own prominent card at the top of the page — |
| kept visually separate from the Behavioral Summary strip below it since it's a |
| different lens (what the model predicted vs. what the feature-derived behavioral |
| signals show).""" |
| if detail is None: |
| msg = ( |
| "This call has no model prediction (not in the out-of-fold test set)." |
| if call_id else "Select a call above to see whether the model got it right." |
| ) |
| return f'<div class="fl-verdict-card"><div class="fl-evidence-empty" style="padding:14px 16px;">{msg}</div></div>' |
|
|
| true_class, predicted_class = detail["true_class"], detail["predicted_class"] |
| correct = predicted_class == true_class |
| banner_cls = "fl-verdict-banner-correct" if correct else "fl-verdict-banner-incorrect" |
| banner_text = "Model Caught It" if correct else "Model Missed It" |
|
|
| def _class_dot(cls: int) -> str: |
| return f'<span style="display:inline-block; width:9px; height:9px; border-radius:50%; background:{CLASS_COLORS[cls]}; flex-shrink:0;"></span>' |
|
|
| return ( |
| '<div class="fl-verdict-card">' |
| f'<div class="fl-verdict-banner {banner_cls}">{banner_text}</div>' |
| '<div class="fl-verdict-tiles">' |
| '<div class="fl-stat-block"><div class="fl-stat-label">Reference Label</div>' |
| f'<div class="fl-stat-value">{_class_dot(true_class)}{CLASS_NAMES[true_class]}</div></div>' |
| '<div class="fl-stat-block"><div class="fl-stat-label">Model Prediction</div>' |
| f'<div class="fl-stat-value">{_class_dot(predicted_class)}{CLASS_NAMES[predicted_class]}</div></div>' |
| '<div class="fl-stat-block"><div class="fl-stat-label">Confidence</div>' |
| f'<div class="fl-stat-value fl-accent">{max(detail["proba"]):.0%}</div></div>' |
| "</div></div>" |
| ) |
|
|
|
|
| def _render_call_view(call_id: str | None, detail: dict | None, selected_turn: int | None = None): |
| """Shared by on_select_call and the random-call auto-load so both produce |
| identically-shaped output — one call selected, optionally one turn pre-highlighted. |
| Verdict card is returned first since it's the punchline, shown above everything |
| else in the Evidence Room.""" |
| if detail is None: |
| empty_msg = '<div class="fl-evidence-empty">This call has no per-turn detail (not in the out-of-fold test set).</div>' |
| return ( |
| render_verdict_card(None, call_id), render_summary_strip(None), [], None, |
| empty_msg, empty_msg, empty_msg, |
| ) |
| turns = detail["turns"] |
| evidence_turn = next((t for t in turns if t["turn_index"] == selected_turn), None) |
| return ( |
| render_verdict_card(detail, call_id), |
| render_summary_strip(detail), |
| turns, |
| selected_turn, |
| render_timeline(turns, selected_turn), |
| render_transcript_synced(turns, selected_turn), |
| render_evidence(evidence_turn), |
| ) |
|
|
|
|
| def _pick_random_call_view( |
| calls_df: pd.DataFrame, |
| ticker: str = "All", |
| cls: str = "All", |
| min_conf: float = 0.0, |
| coverage: str = "OOF-covered only", |
| ): |
| """A fresh random call from the given filters each time this fires — used both by |
| demo.load (fires once per browser session with the default filters, so every new |
| session gets its own pick rather than one baked in at app start-up) and by the |
| sidebar's "Random Call" button (fires with whatever filters are currently active). |
| Prefers OOF-covered calls within that filtered set so the Evidence Room actually |
| has turn-level detail to show; falls back to the full filtered set only if none of |
| it is OOF-covered. Pre-selects the call's first Risk Event turn, if it has one, so |
| the Evidence panel and highlighting are populated immediately rather than waiting |
| for a click.""" |
| pool_df = filter_calls_df(calls_df, ticker, cls, min_conf, coverage) |
| if pool_df.empty: |
| return _render_call_view(None, None) |
| oof_pool = pool_df[pool_df["oof_covered"]] |
| pool = oof_pool if not oof_pool.empty else pool_df |
| call_id = random.choice(pool["call_id"].tolist()) |
| detail = load_call_detail(call_id) |
| turns = detail["turns"] if detail else [] |
| first_risk_turn = next((t["turn_index"] for t in turns if t.get("is_risk_event")), None) |
| return _render_call_view(call_id, detail, selected_turn=first_risk_turn) |
|
|
|
|
| def build_workspace_tab(calls_df: pd.DataFrame, demo: gr.Blocks) -> None: |
| ticker_names = load_ticker_names() |
| ticker_choices = [("All", "All")] + company_dropdown_choices( |
| sorted(calls_df["ticker"].unique().tolist()), ticker_names |
| ) |
|
|
| gr.HTML( |
| '<div style="margin-bottom: 14px;">' |
| '<div class="fl-stat-label" style="margin-bottom: 6px; color: var(--fl-accent);">Listen to the Data</div>' |
| '<div style="color: var(--fl-text-primary); font-size: 15px; line-height: 1.5; max-width: 760px;">' |
| "Below is a real earnings call. The model reads the transcript and listens to " |
| "the executive's vocal delivery — click any highlighted turn to see a moment " |
| "where the two disagreed." |
| "</div></div>" |
| ) |
|
|
| with gr.Row(): |
| with gr.Column(scale=1, elem_classes=["fl-sidebar"]): |
| gr.Markdown("**Calls**") |
| ticker_dd = gr.Dropdown(ticker_choices, value="All", label="Company") |
| class_dd = gr.Dropdown( |
| ["All", "Negative", "Neutral", "Positive"], value="All", label="Sentiment", |
| ) |
| conf_slider = gr.Slider(0, 1, value=0, step=0.05, label="Min. Confidence") |
| random_btn = gr.Button("🎲 Random Call", variant="secondary") |
| calls_table = gr.Dataframe( |
| value=filter_calls(calls_df, "All", "All", 0.0, _DEFAULT_COVERAGE), |
| headers=TABLE_HEADERS, |
| interactive=False, wrap=True, |
| ) |
| filtered_state = gr.State(filter_calls(calls_df, "All", "All", 0.0, _DEFAULT_COVERAGE)) |
|
|
| with gr.Column(scale=4): |
| verdict_html = gr.HTML(render_verdict_card(None, None)) |
| summary_html = gr.HTML(render_summary_strip(None)) |
| gr.Markdown( |
| "*The **Timeline** below tracks four behavioral dimensions per turn — " |
| "**Confidence**, **Stress**, **Specificity**, **Alignment** — split into " |
| "Prepared Remarks and Q&A. Dots mark **Risk Events**, moments where " |
| "friction spiked sharply.*" |
| ) |
| timeline_html = _turn_html_wrapper(render_timeline([], None)) |
| with gr.Row(): |
| with gr.Column(scale=3): |
| gr.Markdown("**Transcript**") |
| transcript_html = _turn_html_wrapper( |
| render_transcript_synced([], None), elem_id="fl-transcript-panel" |
| ) |
| with gr.Column(scale=1): |
| gr.Markdown("**Evidence**") |
| evidence_html = gr.HTML(render_evidence(None)) |
|
|
| turns_state = gr.State([]) |
| selected_turn_state = gr.State(None) |
|
|
| view_outputs = [ |
| verdict_html, summary_html, turns_state, selected_turn_state, timeline_html, transcript_html, |
| evidence_html, |
| ] |
|
|
| def _refresh_calls(ticker, cls, conf): |
| df = filter_calls(calls_df, ticker, cls, conf, _DEFAULT_COVERAGE) |
| return df, df |
|
|
| gr.on( |
| triggers=[ticker_dd.change, class_dd.change, conf_slider.change], |
| fn=_refresh_calls, |
| inputs=[ticker_dd, class_dd, conf_slider], |
| outputs=[calls_table, filtered_state], |
| ) |
|
|
| def on_select_call(evt: gr.SelectData, current_df: pd.DataFrame): |
| if evt.index is None: |
| return gr.update(), gr.update(), [], None, gr.update(), gr.update(), gr.update() |
| row_idx = evt.index[0] |
| call_id = current_df.iloc[row_idx]["call_id"] |
| return _render_call_view(call_id, load_call_detail(call_id)) |
|
|
| calls_table.select(on_select_call, [filtered_state], view_outputs) |
|
|
| demo.load(lambda: _pick_random_call_view(calls_df), outputs=view_outputs) |
|
|
| random_btn.click( |
| lambda ticker, cls, conf: _pick_random_call_view(calls_df, ticker, cls, conf, _DEFAULT_COVERAGE), |
| inputs=[ticker_dd, class_dd, conf_slider], |
| outputs=view_outputs, |
| ) |
|
|
| def on_turn_click(evt: gr.EventData, turns: list[dict]): |
| turn_index = getattr(evt, "turn_index", None) |
| if turn_index is None or not turns: |
| return gr.update(), gr.update(), gr.update(), None |
| turn = next((t for t in turns if t["turn_index"] == turn_index), None) |
| if turn is None: |
| return gr.update(), gr.update(), gr.update(), None |
| return ( |
| render_timeline(turns, turn_index), |
| render_transcript_synced(turns, turn_index), |
| render_evidence(turn), |
| turn_index, |
| ) |
|
|
| timeline_html.click( |
| on_turn_click, [turns_state], [timeline_html, transcript_html, evidence_html, selected_turn_state] |
| ) |
| transcript_html.click( |
| on_turn_click, [turns_state], [timeline_html, transcript_html, evidence_html, selected_turn_state] |
| ) |
|
|
