"""Gradio demo for Agent 9 — Agentic Workflow System."""
from __future__ import annotations
import os
import time
from typing import Any
import gradio as gr
from src.demo_presentation import (
APP_CSS,
EMPTY_ACTIVITY_HTML,
EMPTY_AI_HTML,
EMPTY_JOURNEY_HTML,
EMPTY_OUTCOME_HTML,
EMPTY_PATH_HTML,
EMPTY_ROUTING_HTML,
HERO_HTML,
PATTERN_HTML,
activity_html,
activity_start_html,
playback_delay,
)
from src.demo_workflow import build_onboarding_registry, build_onboarding_workflow
from src.evaluation import run_evaluation_suite
from src.executor import (
WorkflowExecutionError,
WorkflowHumanDecisionError,
execute_workflow,
submit_human_decision,
)
from src.hf_provider import (
build_hf_onboarding_model_from_env,
configured_model_id,
live_ai_enabled_from_env,
)
from src.persistence import RunNotFoundError, SQLiteStateStore
from src.schemas import HumanDecision, WorkflowRun
from src.ui_presenters import (
CASE_HEADERS,
EVENT_HEADERS,
METRIC_HEADERS,
NODE_HEADERS,
REVIEW_HEADERS,
ai_insight_html,
business_journey_html,
business_outcome_html,
evaluation_case_rows,
evaluation_metric_rows,
routing_explanation_html,
run_bundle,
workflow_path_html,
)
WORKFLOW = build_onboarding_workflow()
DATABASE_PATH = os.getenv("WORKFLOW_DB_PATH", "workflow_runs.db")
STATE_STORE = SQLiteStateStore(DATABASE_PATH)
LIVE_MODEL_ID = configured_model_id()
LIVE_ONBOARDING_MODEL = build_hf_onboarding_model_from_env()
STORY_SPECS: dict[str, dict[str, Any]] = {
"Harbor Family — straightforward household": {
"household_id": "HH-HARBOR-STANDARD",
"household_type": "JOINT",
"onboarding_notes": "The fictional Harbor family is beginning a standard advisory relationship. The household has two adults, straightforward planning needs, and a complete synthetic intake.",
"documents_complete": True,
"identity_status": "VERIFIED",
"relationship_complexity": "STANDARD",
"simulation_mode": "NONE",
},
"Harbor Family — temporary onboarding-service issue": {
"household_id": "HH-HARBOR-RETRY",
"household_type": "JOINT",
"onboarding_notes": "The fictional Harbor family has a straightforward synthetic intake, but the demo will simulate one temporary package-preparation service interruption.",
"documents_complete": True,
"identity_status": "VERIFIED",
"relationship_complexity": "STANDARD",
"simulation_mode": "TRANSIENT_ONCE",
},
"Redwood Family Trust — human review required": {
"household_id": "HH-REDWOOD-TRUST",
"household_type": "TRUST",
"onboarding_notes": "The fictional Redwood Family Trust has multiple interested parties and a more complex planning relationship. The synthetic intake is complete but should demonstrate an exception-review path.",
"documents_complete": True,
"identity_status": "VERIFIED",
"relationship_complexity": "COMPLEX",
"simulation_mode": "NONE",
},
"Cedar Household — permanent dependency failure": {
"household_id": "HH-CEDAR-FAILURE",
"household_type": "JOINT",
"onboarding_notes": "The fictional Cedar household has a standard synthetic intake, but the demo will simulate a permanent downstream package-preparation failure.",
"documents_complete": True,
"identity_status": "VERIFIED",
"relationship_complexity": "STANDARD",
"simulation_mode": "PERMANENT",
},
}
RunBundle = tuple[str, str, list[list[Any]], list[list[str]], list[list[str]], dict[str, Any]]
AppBundle = tuple[
str,
str,
list[list[Any]],
list[list[str]],
list[list[str]],
dict[str, Any],
str,
str,
str,
str,
str,
]
def _ai_runtime_html() -> str:
if LIVE_ONBOARDING_MODEL is not None:
return (
'● Live AI connected
'
f'The AI Intake Organizer is using Hugging Face Inference Providers with {LIVE_MODEL_ID}. '
'Its output still must pass the same strict Pydantic schema before the workflow can use it.
'
)
if live_ai_enabled_from_env():
return (
'◐ Live AI requested but not connected
'
'LIVE_AI_ENABLED is on, but no usable HF_TOKEN was found. The demo is using its explicit deterministic fallback.
'
)
return (
'○ AI boundary ready; deterministic fallback active
'
'Live inference is deliberately opt-in. The same bounded node can use Hugging Face Inference Providers when the deployment sets LIVE_AI_ENABLED=true and supplies HF_TOKEN.
'
)
def _build_registry():
return build_onboarding_registry(
onboarding_model=LIVE_ONBOARDING_MODEL,
model_label=LIVE_MODEL_ID if LIVE_ONBOARDING_MODEL is not None else None,
)
def run_request_ui(
household_id: str,
household_type: str,
onboarding_notes: str,
documents_complete: bool,
identity_status: str,
relationship_complexity: str,
simulation_mode: str,
) -> AppBundle:
context = {
"household_id": household_id,
"household_type": household_type,
"onboarding_notes": onboarding_notes,
"documents_complete": documents_complete,
"identity_status": identity_status,
"relationship_complexity": relationship_complexity,
"simulation_mode": simulation_mode,
}
try:
run = execute_workflow(WORKFLOW, _build_registry(), context=context, state_store=STATE_STORE)
return _app_bundle(run)
except WorkflowExecutionError as exc:
return _bundle_with_notice(exc.run, "The onboarding workflow stopped after a controlled failure.")
except Exception as exc:
return _empty_bundle(f"Unable to start onboarding workflow: {type(exc).__name__}")
def run_story_ui(story: str) -> AppBundle:
spec = STORY_SPECS.get(story)
if spec is None:
return _empty_bundle("Choose one of the four onboarding stories before running the workflow.")
return run_request_ui(**spec)
def refresh_run_ui(run_id: str) -> AppBundle:
if not run_id or not run_id.strip():
return _empty_bundle("Enter a run ID to load a persisted workflow.")
try:
return _app_bundle(STATE_STORE.load(run_id.strip()))
except RunNotFoundError:
return _empty_bundle("No persisted workflow run was found for that ID.", run_id.strip())
except Exception as exc:
return _empty_bundle(f"Unable to load workflow: {type(exc).__name__}", run_id.strip())
def submit_human_decision_ui(run_id: str, decision: str) -> AppBundle:
if not run_id or not run_id.strip():
return _empty_bundle("Enter a run ID before submitting a human decision.")
normalized_run_id = run_id.strip()
try:
run = STATE_STORE.load(normalized_run_id)
except RunNotFoundError:
return _empty_bundle("No persisted workflow run was found for that ID.", normalized_run_id)
open_reviews = [review for review in run.human_reviews if review.decision is None]
if len(open_reviews) != 1:
return _bundle_with_notice(run, "This run does not have exactly one open human review. Reload the run and inspect the Human Review table.")
try:
updated = submit_human_decision(
WORKFLOW,
_build_registry(),
run_id=normalized_run_id,
review_id=open_reviews[0].review_id,
decision=HumanDecision(decision),
state_store=STATE_STORE,
)
return _app_bundle(updated)
except (WorkflowHumanDecisionError, WorkflowExecutionError) as exc:
return _bundle_with_notice(exc.run, str(exc))
except ValueError:
return _bundle_with_notice(run, "Unknown human decision.")
except Exception as exc:
return _bundle_with_notice(run, f"Unable to apply decision: {type(exc).__name__}")
def run_evaluation_ui() -> tuple[str, list[list[str]], list[list[str]]]:
report = run_evaluation_suite()
summary = (
"### Wealth Onboarding Workflow Evaluation\n"
f"**Cases:** {report.passed_cases}/{report.total_cases} passed \n"
f"**Pass rate:** {report.case_pass_rate:.2f} \n"
f"**Duplicate executions:** {report.metrics['duplicate_execution_count']} \n"
f"**Invalid transitions:** {report.metrics['invalid_transition_count']}"
)
return summary, evaluation_metric_rows(report), evaluation_case_rows(report)
def _app_bundle(run: WorkflowRun) -> AppBundle:
technical: RunBundle = run_bundle(run)
return (
*technical,
business_outcome_html(run),
workflow_path_html(run),
ai_insight_html(run),
routing_explanation_html(run),
business_journey_html(run),
)
def _bundle_with_notice(run: WorkflowRun, notice: str) -> AppBundle:
technical = run_bundle(run)
return (
f"> {notice}\n\n{technical[0]}",
*technical[1:],
business_outcome_html(run),
workflow_path_html(run),
ai_insight_html(run),
routing_explanation_html(run),
business_journey_html(run),
)
def _empty_bundle(message: str, run_id: str = "") -> AppBundle:
return (
f"### Workflow Status\n{message}", run_id, [], [], [], {},
EMPTY_OUTCOME_HTML, EMPTY_PATH_HTML, EMPTY_AI_HTML, EMPTY_ROUTING_HTML, EMPTY_JOURNEY_HTML,
)
def _running_bundle(run_id: str = "") -> AppBundle:
return (
"### Workflow running…\nThe workflow engine is controlling the approved next steps.",
run_id,
[], [], [], {},
EMPTY_OUTCOME_HTML, EMPTY_PATH_HTML, EMPTY_AI_HTML, EMPTY_ROUTING_HTML, EMPTY_JOURNEY_HTML,
)
def _stream_bundle(bundle: AppBundle):
run_id = bundle[1]
if not run_id:
yield ('STOPPED
Workflow did not start
STOPPED
Workflow trace unavailable