| """Public-safe wealth-management onboarding demo for Agent 9. |
| |
| The fictional Household Onboarding Workflow exercises the workflow runtime end to end |
| with deterministic simulated actions. AI assistance is optional and bounded to intake |
| organization; workflow control, exception routing, retries, persistence, and human |
| approval remain deterministic. All households, notes, and outputs are synthetic. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from typing import Any |
|
|
| from src.model_assist import ModelCall, organize_onboarding_with_model |
| from src.registry import HandlerRegistry |
| from src.retry import RetryableHandlerError |
| from src.schemas import NodeDefinition, NodeType, RetryPolicy, WorkflowDefinition |
|
|
|
|
| ALLOWED_HOUSEHOLD_TYPES = {"INDIVIDUAL", "JOINT", "TRUST", "ENTITY"} |
| ALLOWED_IDENTITY_STATUSES = {"VERIFIED", "REVIEW_REQUIRED"} |
| ALLOWED_COMPLEXITY_LEVELS = {"STANDARD", "COMPLEX"} |
| ALLOWED_SIMULATION_MODES = {"NONE", "TRANSIENT_ONCE", "PERMANENT"} |
| SPECIAL_STRUCTURE_TYPES = {"TRUST", "ENTITY"} |
|
|
|
|
| def build_onboarding_workflow() -> WorkflowDefinition: |
| """Return the fixed fictional household-onboarding workflow used by the public demo.""" |
|
|
| return WorkflowDefinition( |
| workflow_id="wealth-household-onboarding", |
| name="Fictional Wealth Management Household Onboarding", |
| version="1.1", |
| nodes=[ |
| NodeDefinition( |
| node_id="validate_intake", |
| name="Validate Intake", |
| node_type=NodeType.TASK, |
| handler="validate_intake", |
| ), |
| NodeDefinition( |
| node_id="ai_intake_organizer", |
| name="AI Intake Organizer", |
| node_type=NodeType.TASK, |
| handler="ai_intake_organizer", |
| depends_on=["validate_intake"], |
| ), |
| NodeDefinition( |
| node_id="document_check", |
| name="Document Check", |
| node_type=NodeType.TASK, |
| handler="document_check", |
| depends_on=["ai_intake_organizer"], |
| ), |
| NodeDefinition( |
| node_id="policy_check", |
| name="Onboarding Policy Check", |
| node_type=NodeType.TASK, |
| handler="policy_check", |
| depends_on=["document_check"], |
| ), |
| NodeDefinition( |
| node_id="create_onboarding_package", |
| name="Create Onboarding Package", |
| node_type=NodeType.TASK, |
| handler="create_onboarding_package", |
| depends_on=["policy_check", "ai_intake_organizer"], |
| retry_policy=RetryPolicy(retryable=True, max_attempts=3), |
| ), |
| NodeDefinition( |
| node_id="verify_onboarding_package", |
| name="Verify Onboarding Package", |
| node_type=NodeType.TASK, |
| handler="verify_onboarding_package", |
| depends_on=["create_onboarding_package"], |
| ), |
| NodeDefinition( |
| node_id="review_gate", |
| name="Review Gate", |
| node_type=NodeType.DECISION, |
| handler="review_gate", |
| depends_on=["verify_onboarding_package", "policy_check"], |
| routes={ |
| "STANDARD_PATH": "onboarding_ready", |
| "REVIEW_REQUIRED": "human_review", |
| }, |
| ), |
| NodeDefinition( |
| node_id="onboarding_ready", |
| name="Standard Onboarding Ready", |
| node_type=NodeType.TASK, |
| handler="onboarding_ready", |
| depends_on=["review_gate"], |
| ), |
| NodeDefinition( |
| node_id="human_review", |
| name="Operations / Compliance Review", |
| node_type=NodeType.HUMAN_GATE, |
| depends_on=["review_gate"], |
| config={ |
| "reason": ( |
| "A fictional onboarding exception requires explicit operations or " |
| "compliance review before the package can be marked ready." |
| ) |
| }, |
| ), |
| NodeDefinition( |
| node_id="reviewed_onboarding", |
| name="Reviewed Onboarding Ready", |
| node_type=NodeType.TASK, |
| handler="reviewed_onboarding", |
| depends_on=["human_review"], |
| ), |
| ], |
| ) |
|
|
|
|
| def build_onboarding_registry( |
| *, |
| onboarding_model: ModelCall | None = None, |
| model_label: str | None = None, |
| ) -> HandlerRegistry: |
| """Return allowlisted handlers for the fictional onboarding demo. |
| |
| When onboarding_model is omitted, the AI-capable intake step uses a deterministic |
| fallback so the public demo works without credentials. When a model callable is |
| supplied, only the intake-organizer handler delegates to the bounded model adapter; |
| all workflow control and all other handlers remain deterministic. |
| """ |
|
|
| registry = HandlerRegistry() |
| package_attempts: dict[str, int] = {} |
|
|
| def validate_intake(payload: dict[str, Any]) -> dict[str, Any]: |
| context = payload["context"] |
| required_strings = ( |
| "household_id", |
| "household_type", |
| "onboarding_notes", |
| "identity_status", |
| "relationship_complexity", |
| ) |
| normalized: dict[str, Any] = {} |
|
|
| for field in required_strings: |
| value = context.get(field) |
| if not isinstance(value, str) or not value.strip(): |
| raise ValueError(f"{field} must be a non-blank string") |
| normalized[field] = value.strip() |
|
|
| normalized["household_type"] = normalized["household_type"].upper() |
| normalized["identity_status"] = normalized["identity_status"].upper() |
| normalized["relationship_complexity"] = normalized[ |
| "relationship_complexity" |
| ].upper() |
|
|
| if normalized["household_type"] not in ALLOWED_HOUSEHOLD_TYPES: |
| raise ValueError( |
| "household_type must be one of: " |
| + ", ".join(sorted(ALLOWED_HOUSEHOLD_TYPES)) |
| ) |
| if normalized["identity_status"] not in ALLOWED_IDENTITY_STATUSES: |
| raise ValueError( |
| "identity_status must be one of: " |
| + ", ".join(sorted(ALLOWED_IDENTITY_STATUSES)) |
| ) |
| if normalized["relationship_complexity"] not in ALLOWED_COMPLEXITY_LEVELS: |
| raise ValueError( |
| "relationship_complexity must be one of: " |
| + ", ".join(sorted(ALLOWED_COMPLEXITY_LEVELS)) |
| ) |
|
|
| documents_complete = context.get("documents_complete") |
| if not isinstance(documents_complete, bool): |
| raise ValueError("documents_complete must be a boolean") |
| normalized["documents_complete"] = documents_complete |
|
|
| simulation_mode = context.get("simulation_mode", "NONE") |
| if not isinstance(simulation_mode, str): |
| raise ValueError("simulation_mode must be a string") |
| simulation_mode = simulation_mode.strip().upper() |
| if simulation_mode not in ALLOWED_SIMULATION_MODES: |
| raise ValueError( |
| "simulation_mode must be one of: " |
| + ", ".join(sorted(ALLOWED_SIMULATION_MODES)) |
| ) |
| normalized["simulation_mode"] = simulation_mode |
| return normalized |
|
|
| def ai_intake_organizer(payload: dict[str, Any]) -> dict[str, Any]: |
| intake = payload["dependencies"]["validate_intake"] |
|
|
| if onboarding_model is not None: |
| result = organize_onboarding_with_model( |
| onboarding_model, |
| household_type=intake["household_type"], |
| onboarding_notes=intake["onboarding_notes"], |
| ) |
| if model_label: |
| result["model_id"] = model_label |
| return result |
|
|
| household_type = intake["household_type"] |
| if household_type in SPECIAL_STRUCTURE_TYPES: |
| profile_category = "SPECIAL_STRUCTURE" |
| else: |
| profile_category = "STANDARD_HOUSEHOLD" |
| return { |
| "profile_category": profile_category, |
| "summary": ( |
| f"Synthetic {household_type.lower()} household intake organized for " |
| "fictional onboarding review." |
| ), |
| "source": "DETERMINISTIC_FALLBACK", |
| } |
|
|
| def document_check(payload: dict[str, Any]) -> dict[str, Any]: |
| documents_complete = payload["context"]["documents_complete"] |
| return { |
| "documents_complete": documents_complete, |
| "status": "COMPLETE" if documents_complete else "MISSING_ITEMS", |
| } |
|
|
| def policy_check(payload: dict[str, Any]) -> dict[str, Any]: |
| context = payload["context"] |
| documents = payload["dependencies"]["document_check"] |
| exceptions: list[str] = [] |
|
|
| if not documents["documents_complete"]: |
| exceptions.append("MISSING_DOCUMENTS") |
| if str(context["identity_status"]).strip().upper() == "REVIEW_REQUIRED": |
| exceptions.append("IDENTITY_REVIEW_REQUIRED") |
| if str(context["relationship_complexity"]).strip().upper() == "COMPLEX": |
| exceptions.append("COMPLEX_RELATIONSHIP") |
| if str(context["household_type"]).strip().upper() in SPECIAL_STRUCTURE_TYPES: |
| exceptions.append("SPECIAL_STRUCTURE") |
|
|
| return { |
| "eligible_to_prepare_package": True, |
| "exception_reasons": exceptions, |
| } |
|
|
| def create_onboarding_package(payload: dict[str, Any]) -> dict[str, Any]: |
| context = payload["context"] |
| ai_result = payload["dependencies"]["ai_intake_organizer"] |
| household_id = str(context["household_id"]).strip() |
| simulation_mode = str(context.get("simulation_mode", "NONE")).strip().upper() |
| attempt = package_attempts.get(household_id, 0) + 1 |
| package_attempts[household_id] = attempt |
|
|
| if simulation_mode == "TRANSIENT_ONCE" and attempt == 1: |
| raise RetryableHandlerError("synthetic temporary onboarding service interruption") |
| if simulation_mode == "PERMANENT": |
| raise RuntimeError("synthetic permanent onboarding service failure") |
|
|
| return { |
| "package_status": "PREPARED", |
| "service_attempt": attempt, |
| "action": "SIMULATED_ONBOARDING_PACKAGE_PREPARATION", |
| "ai_profile_category": ai_result["profile_category"], |
| "ai_intake_summary": ai_result["summary"], |
| "ai_source": ai_result["source"], |
| } |
|
|
| def verify_onboarding_package(payload: dict[str, Any]) -> dict[str, Any]: |
| package = payload["dependencies"]["create_onboarding_package"] |
| if package.get("package_status") != "PREPARED": |
| raise RuntimeError("onboarding package failed verification") |
| return { |
| "verified": True, |
| "action": package["action"], |
| "ai_summary_included": bool(package.get("ai_intake_summary")), |
| } |
|
|
| def review_gate(payload: dict[str, Any]) -> dict[str, Any]: |
| policy = payload["dependencies"]["policy_check"] |
| reasons = list(policy["exception_reasons"]) |
| return { |
| "route": "REVIEW_REQUIRED" if reasons else "STANDARD_PATH", |
| "exception_reasons": reasons, |
| } |
|
|
| def onboarding_ready(payload: dict[str, Any]) -> dict[str, Any]: |
| decision = payload["dependencies"]["review_gate"] |
| if decision.get("route") != "STANDARD_PATH": |
| raise RuntimeError("standard onboarding finalizer received the wrong route") |
| return { |
| "outcome": "READY_FOR_ADVISOR_REVIEW", |
| "review_path": "STANDARD", |
| } |
|
|
| def reviewed_onboarding(payload: dict[str, Any]) -> dict[str, Any]: |
| review = payload["dependencies"]["human_review"] |
| if review.get("decision") != "APPROVE": |
| raise RuntimeError("reviewed onboarding requires explicit human approval") |
| return { |
| "outcome": "READY_AFTER_HUMAN_REVIEW", |
| "review_path": "EXCEPTION_REVIEW", |
| } |
|
|
| registry.register("validate_intake", validate_intake) |
| registry.register("ai_intake_organizer", ai_intake_organizer) |
| registry.register("document_check", document_check) |
| registry.register("policy_check", policy_check) |
| registry.register("create_onboarding_package", create_onboarding_package) |
| registry.register("verify_onboarding_package", verify_onboarding_package) |
| registry.register("review_gate", review_gate) |
| registry.register("onboarding_ready", onboarding_ready) |
| registry.register("reviewed_onboarding", reviewed_onboarding) |
| return registry |
|
|
|
|
| def example_onboarding( |
| *, |
| household_id: str = "HH-DEMO-001", |
| household_type: str = "JOINT", |
| onboarding_notes: str = ( |
| "Fictional household seeking a standard advisory relationship. " |
| "All intake information in this demo is synthetic." |
| ), |
| documents_complete: bool = True, |
| identity_status: str = "VERIFIED", |
| relationship_complexity: str = "STANDARD", |
| simulation_mode: str = "NONE", |
| ) -> dict[str, Any]: |
| """Return one public-safe synthetic household suitable for demos and tests.""" |
|
|
| return { |
| "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, |
| } |
|
|
|
|
| |
| |
| build_service_request_workflow = build_onboarding_workflow |
| build_service_request_registry = build_onboarding_registry |
| example_request = example_onboarding |
|
|
