Add wellness check-in workflow routing

Add a first-class wellness workflow with burnout signal analysis, trend tracking, and execution routing so non-frontdesk use cases can plug into AgentCare cleanly.

Author: irfanalidv

services/dashboard/app.py

@@ -27,7 +27,7 @@
 from agentcare.doctor import DoctorProfile, load_doctor_schema
 from agentcare.extraction import extract_conversation_fields
 from agentcare.settings import settings
-from agentcare.usecases import process_frontdesk_execution
+from agentcare.usecases import process_agentcare_execution
 
 
 app = FastAPI(title="AgentCare Dashboard", version="0.1.0")
@@ -323,7 +323,7 @@ def _trigger_execution_processing_async(execution_payload: dict[str, Any]) -> bo
 
     def _runner() -> None:
         try:
-            process_frontdesk_execution(
+            process_agentcare_execution(
                 execution_payload,
                 source="dashboard_status_terminal",
                 automate_actions=True,

services/webhooks/app.py

@@ -4,7 +4,7 @@
 from fastapi import FastAPI
 from pydantic import BaseModel, ConfigDict
 
-from agentcare.usecases import process_frontdesk_execution
+from agentcare.usecases import process_agentcare_execution, resolve_execution_workflow
 
 
 app = FastAPI(title="AgentCare Webhooks", version="0.1.0")
@@ -35,21 +35,33 @@ def on_bolna_execution(payload: BolnaExecutionPayload) -> dict[str, Any]:
     and sends confirmation email when customer email + appointment details exist.
     """
     event = payload.model_dump()
-    res = process_frontdesk_execution(
+    workflow = resolve_execution_workflow(event)
+    res = process_agentcare_execution(
         event,
         source="webhook",
+        workflow=workflow,
         automate_actions=True,
         enforce_idempotency=True,
     )
 
     return {
         "ok": res.ok,
-        "deduplicated": res.deduplicated,
+        "workflow": workflow,
+        "deduplicated": getattr(res, "deduplicated", False),
         "execution_id": res.execution_id,
-        "customer_id": res.customer_id,
-        "email_sent": bool(res.email_confirmation),
-        "email_result": res.email_confirmation,
-        "analytics_store": res.analytics_store,
-        "enriched_fields": res.extracted_data,
+        "customer_id": getattr(res, "customer_id", None),
+        "employee_id": getattr(res, "employee_id", None),
+        "email_sent": bool(getattr(res, "email_confirmation", None)),
+        "email_result": getattr(res, "email_confirmation", None),
+        "analytics_store": getattr(res, "analytics_store", None),
+        "enriched_fields": getattr(res, "extracted_data", getattr(res, "extraction", {})),
+        "wellness": {
+            "analysis": getattr(res, "analysis", None),
+            "trend": getattr(res, "trend", None),
+            "policy": getattr(res, "policy", None),
+            "persisted": getattr(res, "persisted", None),
+        }
+        if workflow == "wellness_checkin"
+        else None,
     }
 

src/agentcare/analysis/__init__.py

@@ -1,3 +1,12 @@
+from agentcare.analysis.burnout import BurnoutAnalysis, analyze_burnout_context
 from agentcare.analysis.healthcare import HealthcareAnalysis, analyze_healthcare_context
+from agentcare.analysis.trend import TrendResult, detect_trend
 
-__all__ = ["HealthcareAnalysis", "analyze_healthcare_context"]
+__all__ = [
+    "HealthcareAnalysis",
+    "analyze_healthcare_context",
+    "BurnoutAnalysis",
+    "analyze_burnout_context",
+    "TrendResult",
+    "detect_trend",
+]

src/agentcare/analysis/burnout.py

@@ -0,0 +1,161 @@
+"""
+Burnout signal analysis aligned to Maslach Burnout Inventory dimensions.
+
+This is a signal layer over conversational text. It does not diagnose burnout;
+it produces conservative, evidence-anchored features for wellness routing.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+
+
+@dataclass
+class BurnoutAnalysis:
+    ee_hits: int = 0
+    dp_hits: int = 0
+    pa_hits: int = 0
+    ee_tags: list[str] = field(default_factory=list)
+    dp_tags: list[str] = field(default_factory=list)
+    pa_tags: list[str] = field(default_factory=list)
+    ee_score: float = 0.0
+    dp_score: float = 0.0
+    pa_score: float = 0.0
+    composite_score: float = 0.0
+    risk_band: str = "low"
+    recommended_action: str = "continue_monitoring"
+    high_acuity_flag: bool = False
+
+
+_EE_PATTERNS: list[tuple[str, str]] = [
+    (r"\bexhaust(ed|ing|ion)\b", "exhaustion_explicit"),
+    (r"\bdrained\b", "drained"),
+    (r"\bburn(ed|t)?[\s-]?out\b", "burnt_out_explicit"),
+    (r"\b(can'?t|cannot) (cope|keep up|do this anymore)\b", "cant_cope"),
+    (r"\boverwhelm(ed|ing)\b", "overwhelmed"),
+    (r"\b(too much|too many) (work|things|deadlines|meetings)\b", "workload_excess"),
+    (r"\bworking (late|weekends|all the time|nights)\b", "extended_hours"),
+    (r"\bno (time|energy|break)\b", "no_recovery"),
+    (r"\bnot sleeping\b", "sleep_disruption"),
+    (r"\b(can'?t|cannot) (sleep|relax|switch off|disconnect)\b", "cant_disconnect"),
+    (r"\b(constantly|always) (tired|fatigue)\b", "chronic_fatigue"),
+    (r"\bno energy\b", "low_energy"),
+    (r"\bemotionally (drained|spent|empty)\b", "emotional_depletion"),
+]
+
+_DP_PATTERNS: list[tuple[str, str]] = [
+    (r"\b(don'?t|do not) care anymore\b", "apathy_explicit"),
+    (r"\bjust going through the motions\b", "going_through_motions"),
+    (r"\bcheck(ing|ed) out\b", "checked_out"),
+    (r"\bdisengag(ed|ing)\b", "disengaged"),
+    (r"\bcynical\b", "cynicism_explicit"),
+    (r"\bpointless\b", "pointless"),
+    (r"\b(hate|dread) (work|my job|going in|mondays)\b", "work_aversion"),
+    (r"\bwhat'?s the point\b", "questioning_purpose"),
+    (r"\b(not|no longer) (engaged|invested|excited)\b", "loss_of_engagement"),
+    (r"\b(distant|detached) from (team|colleagues|clients|patients)\b", "interpersonal_detachment"),
+    (r"\b(robotic|mechanical|automatic) at work\b", "mechanical_work"),
+]
+
+_PA_PATTERNS: list[tuple[str, str]] = [
+    (r"\b(not|no longer) (effective|making (a )?difference|productive)\b", "ineffective"),
+    (r"\bnothing (i do )?matters\b", "futility"),
+    (r"\b(failing|failure) at (work|my job)\b", "perceived_failure"),
+    (r"\b(can'?t|cannot) (focus|concentrate|finish|get anything done)\b", "performance_decline"),
+    (r"\b(useless|worthless) at work\b", "self_devaluation"),
+    (r"\bnot good enough\b", "self_doubt"),
+    (r"\b(impostor|imposter)\b", "impostor"),
+    (r"\bmissing deadlines\b", "missed_deadlines"),
+    (r"\b(my )?work quality (has )?(dropped|slipped|declined)\b", "quality_decline"),
+    (r"\b(losing|lost) (confidence|motivation)\b", "motivation_loss"),
+]
+
+_HIGH_ACUITY_PATTERNS: list[tuple[str, str]] = [
+    (r"\bsuicid(al|e)\b", "suicidal_ideation"),
+    (r"\bself[- ]harm\b", "self_harm"),
+    (r"\b(want|wish) to (die|disappear|not (exist|wake up))\b", "wish_to_die"),
+    (r"\bend(ing)? it all\b", "end_it_all"),
+    (r"\bno (way|reason) to (live|go on)\b", "hopelessness_acute"),
+    (r"\b(can'?t|cannot) take (it|this) anymore\b", "breaking_point"),
+    (r"\bhaving a breakdown\b", "breakdown"),
+]
+
+
+def _scan(text: str, patterns: list[tuple[str, str]]) -> tuple[int, list[str]]:
+    hits = 0
+    tags: list[str] = []
+    seen: set[str] = set()
+    for pattern, tag in patterns:
+        if re.search(pattern, text, flags=re.I):
+            hits += 1
+            if tag not in seen:
+                tags.append(tag)
+                seen.add(tag)
+    return hits, tags
+
+
+def _hits_to_score(hits: int) -> float:
+    if hits <= 0:
+        return 0.0
+    return round(min(10.0, 10.0 * (1 - 0.55**hits)), 2)
+
+
+def _fuse(regex_score: float, llm_score: float | None) -> float:
+    if llm_score is None:
+        return regex_score
+    return round(min(10.0, max(0.0, 0.75 * float(llm_score) + 0.25 * regex_score)), 2)
+
+
+def _band(score: float, high_acuity: bool) -> tuple[str, str]:
+    if high_acuity:
+        return "high", "confidential_human_followup_immediate"
+    if score >= 7.0:
+        return "high", "confidential_human_followup"
+    if score >= 4.0:
+        return "medium", "manager_check_in_suggested"
+    return "low", "continue_monitoring"
+
+
+def analyze_burnout_context(
+    *,
+    transcript: str | None,
+    reason: str | None = None,
+    llm_ee: float | None = None,
+    llm_dp: float | None = None,
+    llm_pa: float | None = None,
+) -> BurnoutAnalysis:
+    text = " ".join([reason or "", transcript or ""]).strip().lower()
+    if not text:
+        return BurnoutAnalysis()
+
+    ee_hits, ee_tags = _scan(text, _EE_PATTERNS)
+    dp_hits, dp_tags = _scan(text, _DP_PATTERNS)
+    pa_hits, pa_tags = _scan(text, _PA_PATTERNS)
+    acuity_hits, acuity_tags = _scan(text, _HIGH_ACUITY_PATTERNS)
+
+    ee_score = _fuse(_hits_to_score(ee_hits), llm_ee)
+    dp_score = _fuse(_hits_to_score(dp_hits), llm_dp)
+    pa_score = _fuse(_hits_to_score(pa_hits), llm_pa)
+    composite = round(0.45 * ee_score + 0.35 * dp_score + 0.20 * pa_score, 2)
+
+    high_acuity = acuity_hits > 0
+    band, action = _band(composite, high_acuity)
+    if high_acuity:
+        ee_tags = list(dict.fromkeys(ee_tags + acuity_tags))
+
+    return BurnoutAnalysis(
+        ee_hits=ee_hits,
+        dp_hits=dp_hits,
+        pa_hits=pa_hits,
+        ee_tags=ee_tags[:6],
+        dp_tags=dp_tags[:6],
+        pa_tags=pa_tags[:6],
+        ee_score=ee_score,
+        dp_score=dp_score,
+        pa_score=pa_score,
+        composite_score=composite,
+        risk_band=band,
+        recommended_action=action,
+        high_acuity_flag=high_acuity,
+    )

src/agentcare/analysis/trend.py

@@ -0,0 +1,124 @@
+"""Longitudinal trend detection over per-person wellness score history."""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+
+
+@dataclass
+class TrendResult:
+    n: int
+    slope_per_session: float
+    mk_statistic: float
+    mk_z: float
+    mk_p_value: float
+    direction: str
+    consecutive_deteriorating: int
+    triage_trigger: bool
+
+
+def _ols_slope(values: list[float]) -> float:
+    n = len(values)
+    if n < 2:
+        return 0.0
+    x_mean = (n - 1) / 2.0
+    y_mean = sum(values) / n
+    numerator = 0.0
+    denominator = 0.0
+    for i, value in enumerate(values):
+        dx = i - x_mean
+        numerator += dx * (value - y_mean)
+        denominator += dx * dx
+    return numerator / denominator if denominator else 0.0
+
+
+def _norm_cdf(x: float) -> float:
+    return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))
+
+
+def _mann_kendall(values: list[float]) -> tuple[float, float, float]:
+    n = len(values)
+    if n < 3:
+        return 0.0, 0.0, 1.0
+
+    statistic = 0
+    for i in range(n - 1):
+        for j in range(i + 1, n):
+            if values[j] > values[i]:
+                statistic += 1
+            elif values[j] < values[i]:
+                statistic -= 1
+
+    counts: dict[float, int] = {}
+    for value in values:
+        counts[value] = counts.get(value, 0) + 1
+    ties_term = sum(t * (t - 1) * (2 * t + 5) for t in counts.values() if t > 1)
+    variance = (n * (n - 1) * (2 * n + 5) - ties_term) / 18.0
+    if variance <= 0:
+        return float(statistic), 0.0, 1.0
+
+    if statistic > 0:
+        z_score = (statistic - 1) / math.sqrt(variance)
+    elif statistic < 0:
+        z_score = (statistic + 1) / math.sqrt(variance)
+    else:
+        z_score = 0.0
+
+    p_value = 2.0 * (1.0 - _norm_cdf(abs(z_score)))
+    return float(statistic), float(z_score), float(p_value)
+
+
+def _consecutive_deteriorating(values: list[float]) -> int:
+    if len(values) < 2:
+        return 0
+    run = 0
+    for i in range(len(values) - 1, 0, -1):
+        if values[i] > values[i - 1]:
+            run += 1
+        else:
+            break
+    return run
+
+
+def detect_trend(
+    composite_scores: list[float],
+    *,
+    deterioration_run_threshold: int = 3,
+    score_threshold: float = 7.0,
+    p_value_threshold: float = 0.10,
+) -> TrendResult:
+    n = len(composite_scores)
+    if n == 0:
+        return TrendResult(0, 0.0, 0.0, 0.0, 1.0, "stable", 0, False)
+
+    slope = _ols_slope(composite_scores)
+    statistic, z_score, p_value = _mann_kendall(composite_scores)
+
+    if p_value < p_value_threshold:
+        direction = "deteriorating" if statistic > 0 else "improving" if statistic < 0 else "stable"
+    elif slope >= 0.5:
+        direction = "deteriorating"
+    elif slope <= -0.5:
+        direction = "improving"
+    else:
+        direction = "stable"
+
+    consecutive = _consecutive_deteriorating(composite_scores)
+    latest = composite_scores[-1]
+    triage = (
+        consecutive >= deterioration_run_threshold
+        or latest >= score_threshold
+        or (direction == "deteriorating" and latest >= 5.0)
+    )
+
+    return TrendResult(
+        n=n,
+        slope_per_session=round(slope, 3),
+        mk_statistic=round(statistic, 3),
+        mk_z=round(z_score, 3),
+        mk_p_value=round(p_value, 4),
+        direction=direction,
+        consecutive_deteriorating=consecutive,
+        triage_trigger=triage,
+    )