import html import time import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any from fastapi import HTTPException from fastapi.responses import HTMLResponse from app.state_schema import OnboardingStep @dataclass class LiveOnboardingCase: id: str session_id: str user_id: str employee: dict[str, str] current_step: str = OnboardingStep.WELCOME_SENT pending_signals: list[str] = field(default_factory=lambda: ["document_signed"]) status: str = "waiting_for_employee_signature" document_signed: bool = False hardware_delivered: bool = False adk_status: str = "session_ready" events: list[dict[str, str]] = field(default_factory=list) artifacts: list[dict[str, str]] = field(default_factory=list) created_at: float = field(default_factory=time.time) updated_at: float = field(default_factory=time.time) CASES: dict[str, LiveOnboardingCase] = {} SESSION_TO_CASE: dict[str, str] = {} LATEST_CASE_ID: str | None = None def static_root() -> Path: return Path(__file__).parent / "static" def artifact_root() -> Path: path = Path(__file__).resolve().parents[1] / "local_artifacts" / "onboarding" path.mkdir(parents=True, exist_ok=True) return path def _employee() -> dict[str, str]: return { "name": "Olivia Bennett", "email": "olivia.bennett@example.com", "start_date": "2026-06-01", "role": "Product Manager", "team": "Platform Systems", "manager": "Avery Stone", "corporate_email": "olivia.bennett@example.com", "tracking_id": "HW-55443", "photo_url": "/live-onboarding/olivia-bennett.jpg", } def _case_dir(case_id: str) -> Path: path = artifact_root() / case_id path.mkdir(parents=True, exist_ok=True) return path def _event(case: LiveOnboardingCase, kind: str, title: str, detail: str) -> None: case.events.insert( 0, { "kind": kind, "title": title, "detail": detail, "time": time.strftime("%H:%M:%S"), }, ) case.updated_at = time.time() def _artifact( case: LiveOnboardingCase, artifact_id: str, title: str, kind: str, filename: str ) -> None: href = f"/api/live-onboarding/cases/{case.id}/artifacts/{artifact_id}" existing = next( (item for item in case.artifacts if item["id"] == artifact_id), None ) payload = { "id": artifact_id, "title": title, "kind": kind, "filename": filename, "href": href, "created_at": time.strftime("%H:%M:%S"), } if existing: existing.update(payload) else: case.artifacts.insert(0, payload) def _packet_html(case: LiveOnboardingCase, signed: bool = False) -> str: employee = case.employee signature = ( html.escape(employee["name"]) if signed else "Pending employee signature" ) signed_at = time.strftime("%Y-%m-%d %H:%M:%S %Z") if signed else "" signature_class = "signed" if signed else "pending" return f""" Onboarding Packet - {html.escape(employee["name"])}

New Hire Onboarding Packet

This local packet is generated for the demo workflow. It is not a legal contract or e-signature integration.

Packet ID: {html.escape(case.id)}
Session ID: {html.escape(case.session_id)}
Generated locally
Employee{html.escape(employee["name"])}
Email{html.escape(employee["email"])}
Role{html.escape(employee["role"])}
Start date{html.escape(employee["start_date"])}
Team{html.escape(employee["team"])}
Manager{html.escape(employee["manager"])}

Welcome

Welcome to Platform Systems. This packet confirms the onboarding details that the HR onboarding coordinator will use to prepare access, equipment, and the Day One schedule.

Employee acknowledgements

Signature {signature} {"Signed locally at " + html.escape(signed_at) + "" if signed else ""}
""" def _schedule_html(case: LiveOnboardingCase) -> str: employee = case.employee return f""" Day One Schedule

Day One Schedule for {html.escape(employee["name"])}

Sent to {html.escape(employee["corporate_email"])} after hardware delivery was confirmed.

  1. 09:00Welcome and IT login setup
  2. 10:00Meet {html.escape(employee["manager"])} and the team
  3. 11:30Platform Systems overview
  4. 14:00Security and device walkthrough
""" def _hardware_receipt_html(case: LiveOnboardingCase) -> str: employee = case.employee return f""" Hardware Delivery Receipt

Hardware Delivery Receipt

Employee-confirmed local receipt generated by the onboarding demo.

Confirmed delivery

{html.escape(employee["tracking_id"])} delivered to {html.escape(employee["name"])}.

""" def _write_artifact(case: LiveOnboardingCase, filename: str, body: str) -> None: (_case_dir(case.id) / filename).write_text(body, encoding="utf-8") def case_payload(case: LiveOnboardingCase) -> dict[str, Any]: return { "id": case.id, "session_id": case.session_id, "user_id": case.user_id, "employee": case.employee, "current_step": case.current_step, "pending_signals": case.pending_signals, "status": case.status, "document_signed": case.document_signed, "hardware_delivered": case.hardware_delivered, "adk_status": case.adk_status, "events": case.events, "artifacts": case.artifacts, "updated_at": case.updated_at, } async def create_live_case(session_service) -> LiveOnboardingCase: global LATEST_CASE_ID employee = _employee() case_id = str(uuid.uuid4()) case = LiveOnboardingCase( id=case_id, session_id=case_id, user_id="employee", employee=employee, ) await session_service.create_session( app_name="app", user_id=case.user_id, session_id=case.session_id, state={ "current_step": OnboardingStep.WELCOME_SENT, "new_hire_details": { "name": employee["name"], "email": employee["email"], "start_date": employee["start_date"], }, "pending_signals": ["document_signed"], }, ) _write_artifact(case, "welcome_packet.html", _packet_html(case)) _artifact( case, "welcome-packet", "Unsigned onboarding packet", "html", "welcome_packet.html", ) _event( case, "agent", "Welcome packet generated", "A real local HTML onboarding packet was generated and attached to this case.", ) _event( case, "state", "Agent is waiting", "ADK session is parked at WELCOME_SENT until the employee signs the packet.", ) CASES[case.id] = case SESSION_TO_CASE[case.session_id] = case.id LATEST_CASE_ID = case.id return case async def mark_document_signed(case: LiveOnboardingCase, resume_handler) -> None: case.document_signed = True case.status = "waking_after_signature" case.current_step = OnboardingStep.DOCUMENTS_SIGNED case.pending_signals = [] _write_artifact( case, "signed_onboarding_packet.html", _packet_html(case, signed=True) ) _artifact( case, "signed-packet", "Signed onboarding packet", "html", "signed_onboarding_packet.html", ) _event( case, "employee", f"{case.employee['name']} signed the packet", "The employee clicked Sign Packet. A signed local artifact was stored.", ) _event( case, "webhook", "document_signed webhook fired", "The app called the same ADK resume handler used by /webhooks/document_signed.", ) try: await resume_handler.receive_signed_documents_callback( user_id=case.user_id, session_id=case.session_id ) case.adk_status = "document_resume_completed" case.current_step = OnboardingStep.IT_PROVISIONED case.pending_signals = ["hardware_delivered"] case.status = "waiting_for_hardware_delivery" _event( case, "agent", "ADK resume completed", "The document signature wake turn completed. HR can now wait for hardware delivery.", ) except Exception as exc: case.adk_status = "document_resume_failed" case.status = "document_signed_adk_resume_failed" _event(case, "error", "ADK resume failed", str(exc)) finally: case.updated_at = time.time() async def mark_hardware_delivered(case: LiveOnboardingCase, resume_handler) -> None: case.hardware_delivered = True case.status = "waking_after_hardware_delivery" case.current_step = OnboardingStep.HARDWARE_DELIVERED case.pending_signals = [] _write_artifact( case, "hardware_delivery_receipt.html", _hardware_receipt_html(case), ) _artifact( case, "hardware-receipt", "Hardware delivery receipt", "html", "hardware_delivery_receipt.html", ) _event( case, "employee", f"{case.employee['name']} confirmed laptop delivery", f"Employee confirmed receipt of tracking ID {case.employee['tracking_id']}.", ) _event( case, "webhook", "hardware_delivered webhook fired", "The app called the same ADK resume handler used by /webhooks/hardware_delivered.", ) try: await resume_handler.receive_hardware_delivery_callback( user_id=case.user_id, session_id=case.session_id, tracking_id=case.employee["tracking_id"], ) case.adk_status = "hardware_resume_completed" case.current_step = OnboardingStep.COMPLETED case.status = "completed" _write_artifact(case, "day_one_schedule.html", _schedule_html(case)) _artifact( case, "day-one-schedule", "Day One schedule", "html", "day_one_schedule.html", ) _event( case, "agent", "Onboarding completed", "The hardware wake turn completed and a local Day One schedule artifact was stored.", ) except Exception as exc: case.adk_status = "hardware_resume_failed" case.status = "hardware_delivered_adk_resume_failed" _event(case, "error", "ADK resume failed", str(exc)) finally: case.updated_at = time.time() def case_for_session(session_id: str) -> LiveOnboardingCase | None: case_id = SESSION_TO_CASE.get(session_id) if not case_id: return None return CASES.get(case_id) def get_case(case_id: str) -> LiveOnboardingCase: case = CASES.get(case_id) if not case: raise HTTPException(status_code=404, detail="Onboarding case not found") return case def artifact_response(case_id: str, artifact_id: str) -> HTMLResponse: case = get_case(case_id) artifact = next( (item for item in case.artifacts if item["id"] == artifact_id), None ) if not artifact: raise HTTPException(status_code=404, detail="Artifact not found") path = _case_dir(case.id) / artifact["filename"] if not path.exists(): raise HTTPException(status_code=404, detail="Artifact file not found") return HTMLResponse(path.read_text(encoding="utf-8")) def empty_case_payload() -> dict[str, Any]: return {"active": False, "message": "No live onboarding case has been started."} def latest_case_payload() -> dict[str, Any]: if not LATEST_CASE_ID or LATEST_CASE_ID not in CASES: return empty_case_payload() return {"active": True, "case": case_payload(CASES[LATEST_CASE_ID])}