"""Presentation helpers for the Agent 9 Gradio demo. These helpers translate workflow and evaluation objects into display-safe views. The business presenters explain the fictional wealth-onboarding workflow in plain English, while the technical presenters preserve node, review, and audit evidence. Workflow context is never copied into the audit timeline. """ from __future__ import annotations import html import json from typing import Any from src.evaluation import EvaluationReport from src.schemas import WorkflowRun, WorkflowStatus NODE_HEADERS = ["Node", "Status", "Attempt", "Error", "Output"] EVENT_HEADERS = ["Timestamp (UTC)", "Event", "Node", "Details"] REVIEW_HEADERS = [ "Review ID", "Node", "Decision", "Reason", "Requested At (UTC)", "Decided At (UTC)", ] METRIC_HEADERS = ["Metric", "Value"] CASE_HEADERS = ["Case", "Category", "Passed", "Expected", "Observed"] BUSINESS_STEPS = [ ("validate_intake", "Check the household intake", "Make sure the fictional intake is complete and usable."), ("ai_intake_organizer", "🤖 AI Intake Organizer", "Organize synthetic notes into a bounded household profile and summary."), ("document_check", "Check onboarding documents", "Record whether the fictional document set is complete."), ("policy_check", "Apply onboarding rules", "Identify deterministic exception reasons that require extra review."), ("create_onboarding_package", "Prepare the onboarding package", "Create a simulated package that includes the bounded AI work product."), ("verify_onboarding_package", "Verify the package", "Confirm the simulated onboarding package was prepared correctly."), ("review_gate", "Choose the review path", "Route only between the approved standard path and human-review path."), ("onboarding_ready", "Standard package ready", "Mark a straightforward fictional household ready for advisor review."), ("human_review", "Operations / compliance review", "Pause exception cases until an explicit person decides."), ("reviewed_onboarding", "Ready after human review", "Continue only after an exception case is explicitly approved."), ] _EXCEPTION_LABELS = { "MISSING_DOCUMENTS": "Missing onboarding documents", "IDENTITY_REVIEW_REQUIRED": "Identity information needs review", "COMPLEX_RELATIONSHIP": "Relationship marked complex", "SPECIAL_STRUCTURE": "Trust or entity structure", } def run_summary(run: WorkflowRun) -> str: """Return a concise technical Markdown summary for one workflow run.""" completed = sum(node.status.value == "COMPLETED" for node in run.node_runs.values()) skipped = sum(node.status.value == "SKIPPED" for node in run.node_runs.values()) failed = sum(node.status.value == "FAILED" for node in run.node_runs.values()) waiting_reviews = sum(review.decision is None for review in run.human_reviews) lines = [ f"### Run `{run.run_id}`", f"**Workflow:** `{run.workflow_id}` v`{run.workflow_version}` ", f"**Status:** `{run.status.value}` ", f"**Nodes:** {completed} completed · {skipped} skipped · {failed} failed ", f"**Events:** {len(run.events)} · **Open human reviews:** {waiting_reviews}", ] if run.status is WorkflowStatus.WAITING_FOR_HUMAN: lines.append("\n**Action required:** an explicit operations/compliance decision is required before the workflow can continue.") elif run.status is WorkflowStatus.REJECTED: lines.append("\nThe fictional onboarding process was rejected by a human reviewer; this is not a technical failure.") elif run.status is WorkflowStatus.FAILED: lines.append("\nThe onboarding workflow stopped safely after a controlled failure.") return "\n".join(lines) def business_outcome_html(run: WorkflowRun) -> str: """Explain the current onboarding outcome without requiring workflow terminology.""" review_route = _review_route(run) attempts = _package_attempts(run) if run.status is WorkflowStatus.WAITING_FOR_HUMAN: tone = "waiting" eyebrow = "HUMAN CHECKPOINT" title = "An onboarding exception needs a person" body = ( "The routine preparation steps are complete, but one or more deterministic " "exception rules require operations or compliance review. Automation is paused " "and cannot mark the package ready on its own." ) takeaway = "Pattern shown: AI contributes intake work product, but people retain authority over exception decisions." elif run.status is WorkflowStatus.REJECTED: tone = "rejected" eyebrow = "BUSINESS DECISION" title = "The reviewer stopped this onboarding case" body = ( "A person rejected the fictional exception case. The workflow ended intentionally; " "nothing crashed and the reviewed-ready step did not run." ) takeaway = "Pattern shown: a valid human business decision is different from a technical failure." elif run.status is WorkflowStatus.FAILED: tone = "failed" eyebrow = "SAFE FAILURE" title = "The onboarding workflow stopped safely" body = ( "A permanent simulated dependency problem occurred. Instead of continuing with an " "uncertain package, the workflow contained the failure and blocked later steps." ) takeaway = "Pattern shown: failures are contained instead of cascading through an onboarding process." elif run.status is WorkflowStatus.COMPLETED and review_route == "REVIEW_REQUIRED": tone = "success" eyebrow = "HUMAN-GUIDED COMPLETION" title = "The onboarding package is ready after human review" body = ( "Automation prepared and verified the fictional package, paused at the exception gate, " "and resumed only after an explicit human approval." ) takeaway = "Pattern shown: automation handles the process while a person retains authority over exceptions." elif run.status is WorkflowStatus.COMPLETED and attempts > 1: tone = "success" eyebrow = "AUTOMATIC RECOVERY" title = "A temporary onboarding-service problem occurred — and recovered" body = ( f"Package preparation needed {attempts} attempts. The bounded retry policy handled " "the temporary interruption and the fictional onboarding package still reached the standard ready path." ) takeaway = "Pattern shown: temporary operational failures can recover without an endless retry loop." elif run.status is WorkflowStatus.COMPLETED: tone = "success" eyebrow = "STRAIGHTFORWARD ONBOARDING" title = "The fictional household package is ready for advisor review" body = ( "The intake was valid, the package was prepared and verified, and no deterministic " "exception rule required an additional operations/compliance checkpoint." ) takeaway = "Pattern shown: AI helps prepare the work product while deterministic rules control consequences." else: tone = "neutral" eyebrow = "WORKFLOW IN PROGRESS" title = "The onboarding workflow is processing the household" body = "Application code is controlling the order of work, state changes, and approved next steps." takeaway = "Pattern shown: AI assists inside a bounded step; it does not improvise the onboarding process." return ( f'
' f'
{html.escape(eyebrow)}
' f'
{html.escape(title)}
' f'
{html.escape(body)}
' f'
{html.escape(takeaway)}
' "
" ) def ai_insight_html(run: WorkflowRun) -> str: """Show the bounded AI work product and clearly distinguish live vs fallback mode.""" node = run.node_runs.get("ai_intake_organizer") if node is None or not isinstance(node.output, dict): if node is not None and node.status.value == "FAILED": return ( '
' '
AI INTAKE ORGANIZER
' '
The AI boundary failed safely
' '
No unvalidated model output was allowed into the onboarding package or routing logic.
' '
' ) return ( '
' '
AI INTAKE ORGANIZER
' '
AI work product not available yet
' '
Run a fictional household to see the bounded profile category and intake summary.
' '
' ) source = str(node.output.get("source", "UNKNOWN")) category = str(node.output.get("profile_category", "UNKNOWN")) summary = str(node.output.get("summary", "")) model_id = str(node.output.get("model_id", "")) live = source == "MODEL_ASSISTED" mode_label = "LIVE HUGGING FACE MODEL" if live else "DETERMINISTIC FALLBACK" model_line = f'
Model: {html.escape(model_id)}
' if model_id else "" included = _ai_summary_included(run) included_text = ( "The validated AI summary was included in the simulated onboarding package." if included else "The AI work product has not yet reached the package-preparation step." ) return ( f'
' '
🤖 BOUNDED AI WORK PRODUCT
' f'
{html.escape(mode_label)}
' f'{model_line}' f'
Profile category: {html.escape(category)}
' f'
Intake summary: {html.escape(summary)}
' f'
✓ {html.escape(included_text)}
' '
AI did not decide document completeness, exception rules, routing, retries, or approval.
' '
' ) def routing_explanation_html(run: WorkflowRun) -> str: """Explain exactly why deterministic code selected the current branch.""" gate = run.node_runs.get("review_gate") if gate is None or not isinstance(gate.output, dict): return ( '
' '
WHY DID IT ROUTE HERE?
' '
No routing decision has been made yet
' '
The workflow must finish package preparation and verification before the deterministic review gate can choose a branch.
' '
' ) route = str(gate.output.get("route", "")) reasons = [str(reason) for reason in gate.output.get("exception_reasons") or []] if route == "STANDARD_PATH": return ( '
' '
WHY DID IT ROUTE HERE?
' '
Standard path selected by code
' '
The deterministic onboarding rules found no configured exception trigger. No human checkpoint was required for this fictional preparation flow.
' '
✓ The AI summary did not choose this route.
' '
' ) items = "".join( f'
  • {html.escape(_EXCEPTION_LABELS.get(reason, reason.replace("_", " ").title()))}
    • ' for reason in reasons ) return ( '
    ' '
    WHY DID IT ROUTE HERE?
    ' '
    Human review required by deterministic rules
    ' '
    The workflow found these configured exception conditions:
    ' f'' '
    ✓ These reasons come from structured inputs and code—not from the AI-generated summary.
    ' '
    ' ) def workflow_path_html(run: WorkflowRun) -> str: """Render a compact visual DAG with the actual path reflected by node state.""" main_nodes = [ ("validate_intake", "Validate"), ("ai_intake_organizer", "🤖 AI organize"), ("document_check", "Documents"), ("policy_check", "Rules"), ("create_onboarding_package", "Prepare package"), ("verify_onboarding_package", "Verify"), ("review_gate", "Review gate"), ] main_html = _flow_nodes(run, main_nodes) standard_html = _flow_nodes(run, [("onboarding_ready", "✅ Standard ready")]) review_html = _flow_nodes( run, [ ("human_review", "👤 Human review"), ("reviewed_onboarding", "✅ Ready after review"), ], ) return ( '
    ' '
    Live workflow path
    ' '
    Completed steps light up; the unused branch is muted. The blue-outlined node is the only AI-assisted step.
    ' f'
    {main_html}
    ' '
    ' f'
    STANDARD PATH
    {standard_html}
    ' f'
    EXCEPTION PATH
    {review_html}
    ' '
    ' '
    Blue outline = AI boundary · Green = completed · Amber = waiting · Red = stopped · Gray = not selected/not reached
    ' '
    ' ) def business_journey_html(run: WorkflowRun) -> str: """Render technical node state as an understandable onboarding journey.""" cards: list[str] = [] for node_id, title, description in BUSINESS_STEPS: icon, state_label, tone = _friendly_node_state(run, node_id) detail = _friendly_node_detail(run, node_id) cards.append( '
    ' f'
    {icon}
    ' '
    ' f'
    {html.escape(title)}
    ' f'
    {html.escape(state_label)}
    ' f'
    {html.escape(detail or description)}
    ' "
    " ) return ( '
    ' '
    What happened, step by step
    ' '
    The AI step creates useful intake work product. Deterministic rules decide whether the fictional household stays on the standard path or pauses for a person.
    ' f'
    {"".join(cards)}
    ' "
    " ) def node_rows(run: WorkflowRun) -> list[list[Any]]: """Return display rows for node runtime state without workflow context.""" rows: list[list[Any]] = [] for node_id, node in run.node_runs.items(): rows.append( [ node_id, node.status.value, node.attempt, node.error or "", _json_text(node.output), ] ) return rows def event_rows(run: WorkflowRun) -> list[list[str]]: """Return append-only audit events as control-metadata rows.""" return [ [ event.timestamp.isoformat(), event.event_type.value, event.node_id or "", _json_text(event.details), ] for event in run.events ] def review_rows(run: WorkflowRun) -> list[list[str]]: """Return structured human-review history for display.""" return [ [ review.review_id, review.node_id, review.decision.value if review.decision is not None else "OPEN", review.reason, review.requested_at.isoformat(), review.decided_at.isoformat() if review.decided_at is not None else "", ] for review in run.human_reviews ] def evaluation_metric_rows(report: EvaluationReport) -> list[list[str]]: """Return sorted evaluation metrics for a compact UI table.""" rows = [["total_cases", str(report.total_cases)]] rows.extend( [metric_name, _metric_text(metric_value)] for metric_name, metric_value in sorted(report.metrics.items()) ) return rows def evaluation_case_rows(report: EvaluationReport) -> list[list[str]]: """Return readable evaluation case summaries without raw test payloads.""" return [ [ case.case_id, case.category, "PASS" if case.passed else "FAIL", case.expected, case.observed, ] for case in report.cases ] def run_bundle(run: WorkflowRun) -> tuple[ str, str, list[list[Any]], list[list[str]], list[list[str]], dict[str, Any], ]: """Return the standard technical output bundle consumed by Gradio callbacks.""" return ( run_summary(run), run.run_id, node_rows(run), event_rows(run), review_rows(run), run.final_output or {}, ) def _review_route(run: WorkflowRun) -> str | None: node = run.node_runs.get("review_gate") if node is None or not isinstance(node.output, dict): return None route = node.output.get("route") return str(route) if route is not None else None def _package_attempts(run: WorkflowRun) -> int: node = run.node_runs.get("create_onboarding_package") return node.attempt if node is not None else 0 def _ai_summary_included(run: WorkflowRun) -> bool: node = run.node_runs.get("verify_onboarding_package") if node is None or not isinstance(node.output, dict): return False return bool(node.output.get("ai_summary_included")) def _latest_review_decision(run: WorkflowRun) -> str | None: if not run.human_reviews: return None decision = run.human_reviews[-1].decision return decision.value if decision is not None else None def _friendly_node_state(run: WorkflowRun, node_id: str) -> tuple[str, str, str]: node = run.node_runs.get(node_id) if node is None: return "•", "Not reached", "pending" status = node.status.value if status == "COMPLETED": if node_id == "human_review": decision = _latest_review_decision(run) if decision == "APPROVE": return "✓", "Approved by a person", "done" if decision == "REJECT": return "■", "Rejected by a person", "rejected" if node_id == "create_onboarding_package" and node.attempt > 1: return "↻", f"Recovered on attempt {node.attempt}", "done" return "✓", "Done", "done" if status == "WAITING_FOR_HUMAN": return "!", "Waiting for a person", "waiting" if status == "SKIPPED": return "↷", "Not needed on this path", "skipped" if status == "FAILED": return "×", "Stopped safely here", "failed" if status == "RETRY_SCHEDULED": return "↻", "Retry scheduled", "waiting" if status == "RUNNING": return "…", "In progress", "waiting" return "•", "Not reached", "pending" def _friendly_node_detail(run: WorkflowRun, node_id: str) -> str | None: node = run.node_runs.get(node_id) if node is None: return None if node_id == "ai_intake_organizer" and isinstance(node.output, dict): source = node.output.get("source") category = node.output.get("profile_category") if source == "MODEL_ASSISTED": return f"A live bounded model organized the intake and proposed {category}; its validated summary was available to package preparation, not routing." if source == "DETERMINISTIC_FALLBACK": return f"This is the AI-capable boundary. The deterministic fallback produced {category}; the same validated output shape feeds package preparation." if node_id == "create_onboarding_package" and node.status.value == "COMPLETED": if node.attempt > 1: return f"A temporary service problem occurred, but the bounded retry policy recovered in {node.attempt} attempts. The validated AI intake summary was retained as package work product." return "The fictional onboarding package was prepared on the first attempt and included the validated AI intake work product." if node_id == "review_gate" and isinstance(node.output, dict): route = node.output.get("route") reasons = node.output.get("exception_reasons") or [] if route == "REVIEW_REQUIRED": friendly = [_EXCEPTION_LABELS.get(str(reason), str(reason).replace("_", " ").lower()) for reason in reasons] reason_text = ", ".join(friendly) if friendly else "an onboarding exception" return f"Deterministic rules found {reason_text}, so the standard-ready path was blocked." if route == "STANDARD_PATH": return "Deterministic rules found no exception requiring the extra human-review path." if node_id == "human_review": decision = _latest_review_decision(run) if node.status.value == "WAITING_FOR_HUMAN": return "The workflow is deliberately paused and cannot mark the package ready on its own." if decision == "APPROVE": return "A person explicitly approved the exception case, allowing the workflow to resume." if decision == "REJECT": return "A person explicitly rejected the exception case, ending the fictional onboarding process." if decision == "RETRY": return "A person requested another review cycle without replaying upstream work." if node_id == "onboarding_ready" and node.status.value == "COMPLETED": return "The straightforward fictional package is ready for the next advisor-facing step; no account is actually opened." if node_id == "reviewed_onboarding" and node.status.value == "COMPLETED": return "The exception package is ready only because a person explicitly approved it; no account is actually opened." if node.status.value == "SKIPPED": return "This step belongs to the other approved branch, so it was intentionally skipped." if node.status.value == "FAILED": return "The workflow contained the problem here and did not continue into later onboarding work." return None def _flow_nodes(run: WorkflowRun, nodes: list[tuple[str, str]]) -> str: parts: list[str] = [] for index, (node_id, label) in enumerate(nodes): if index: parts.append('→') parts.append(_map_node(run, node_id, label)) return "".join(parts) def _map_node(run: WorkflowRun, node_id: str, label: str) -> str: node = run.node_runs.get(node_id) state = node.status.value.lower().replace("_", "-") if node is not None else "pending" ai_class = " map-node-ai" if node_id == "ai_intake_organizer" else "" return ( f'{html.escape(label)}' ) def _json_text(value: Any) -> str: if value is None: return "" return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str) def _metric_text(value: float | int) -> str: if isinstance(value, float): return f"{value:.2f}" return str(value)