Multi-Tenancy Phase 2 — Risk Assessment + MSO Second Opinion¶

consent_given
    ↓
risk_review_pending ← [Rule engine scores, human reviews]
    ↓
risk_cleared
    ↓
providers_notified → quoting → quotes_pooled → patient_reviewing → provider_selected
    ↓
mso_offered ← [Patient chooses: document review, video, or skip]
    ↓
mso_complete (or decline_mso → payment_locked)
    ↓
payment_locked

NOTE: State machine needs TWO skip paths:
  - skip_mso: provider_selected → payment_locked (patient never enters mso_offered)
  - decline_mso: mso_offered → payment_locked (patient entered but declined)
  Both are pre-existing gaps — add decline_mso transition in case_machine.py.

@dataclass
class RiskAssessmentResult:
    """Typed return from risk scoring — never a raw dict."""
    risk_score: int           # 0-100
    risk_level: str           # low | moderate | high
    blocking_count: int
    risk_factors: list[dict]
    auto_cleared: bool
    review_required: bool


class RiskScoringService:
    """Scores risk but does NOT mutate case state.

    DESIGN: This is a domain service, not an orchestrator. It returns
    RiskAssessmentResult and the CALLER (case lifecycle service or router)
    is responsible for state transitions. This avoids cross-domain session sharing.
    """

    async def assess_case(
        self, db: AsyncSession, case_id: str, tenant_id: str
    ) -> RiskAssessmentResult:
        """Run risk assessment on a case.

        1. Load EHR snapshot
        2. Call risk_assessor.assess_risks(ehr)
        3. Compute aggregate risk score (0-100)
        4. Determine auto-clear eligibility
        5. Store results on case (risk_assessment JSONB)

        Does NOT transition case state — caller handles that.
        """

    def compute_risk_score(self, risk_factors: list[dict]) -> int:
        """Aggregate risk factors into a 0-100 score.

        Scoring:
        - Each high severity: +20
        - Each moderate: +10
        - Each low: +3
        - Each blocking: +25 (additive with severity)
        - Cap at 100
        """

    def determine_auto_clear(self, risk_score: int, blocking_count: int) -> bool:
        """Low-risk cases with no blocking factors can auto-clear.

        Requires risk_review_human_required=false flag.
        """

risk_assessment: Mapped[dict | None] = mapped_column(FlexibleJSON, nullable=True)
# Stores: { risk_score, risk_level, risk_factors, assessed_at, assessed_by }

ALTER TABLE cases ADD COLUMN risk_assessment JSONB;

GET  /api/v1/risk/queue
  → Cases in risk_review_pending, sorted by risk_score desc
  → Requires: @require_permission("case:assign")

GET  /api/v1/risk/{case_id}/assessment
  → Risk factors + EHR snapshot + patient context
  → Requires: @require_permission("case:read:assigned")

POST /api/v1/risk/{case_id}/decision
  Body: { decision: "approved|rejected|more_info", notes, override_blocking: bool, override_reason }
  → Transitions case state + audit log
  → If approved: risk_review_pending → risk_cleared
  → If more_info: stays in risk_review_pending, patient notified
  → If rejected: case → closed with reason
  → Requires: @require_permission("case:assign")

GET  /api/v1/risk/{case_id}/history
  → Audit trail of risk decisions

# Stored in case.risk_assessment after review:
{
    "risk_score": 35,
    "risk_level": "moderate",
    "risk_factors": [...],
    "assessed_at": "2026-04-19T...",
    "review_decision": "approved",
    "reviewed_by": "coordinator-clerk-id",
    "reviewed_at": "2026-04-19T...",
    "review_notes": "Low risk, no blocking factors",
    "override_blocking": false,
    "override_reason": null
}

# MSO credentialing
mso_credentialing_status: Mapped[str | None] = mapped_column(String(30))
# pending_review | verified | suspended | inactive
mso_verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))

# Consultation rates (USD cents)
consultation_hourly_rate_cents: Mapped[int | None] = mapped_column(Integer)
document_review_rate_cents: Mapped[int | None] = mapped_column(Integer)

# Availability
max_concurrent_consultations: Mapped[int] = mapped_column(Integer, default=3)
consultation_languages: Mapped[list | None] = mapped_column(FlexibleJSON)

ALTER TABLE doctors
    ADD COLUMN mso_credentialing_status VARCHAR(30),
    ADD COLUMN mso_verified_at TIMESTAMPTZ,
    ADD COLUMN consultation_hourly_rate_cents INTEGER,
    ADD COLUMN document_review_rate_cents INTEGER,
    ADD COLUMN max_concurrent_consultations INTEGER DEFAULT 3;
-- NOTE: consultation_languages already exists on doctors table — do NOT re-add

class MSOMatchingService:

    async def match_mso_doctors(
        self, db: AsyncSession, *,
        case_id: str,
        procedure_code: str,
        patient_language: str,
        selected_provider_id: str,  # Exclude same-hospital doctors
        consultation_format: str,  # document_review | video | both
    ) -> list[dict]:
        """Find top 3 MSO doctors for a case.

        Ranking factors:
        - Specialty match (procedure ↔ doctor specialty): 0.35
        - Language concordance: 0.25
        - Credentials (board certs): 0.20
        - Availability + response time: 0.15
        - Patient ratings: 0.05

        Exclusions:
        - Doctors affiliated with selected_provider_id
        - mso_credentialing_status != 'verified'
        - is_active != true
        """

case_id: Mapped[str | None] = mapped_column(String(36), ForeignKey("cases.id"))
doctor_id: Mapped[str | None] = mapped_column(String(36), ForeignKey("doctors.id"))
# The MSO specialist — distinct from provider_id (hospital). Both can be set:
# provider_id = the hospital context, doctor_id = the individual MSO doctor.
consultation_format: Mapped[str | None] = mapped_column(String(30))
# document_review | video_call
cost_cents: Mapped[int | None] = mapped_column(Integer)
cost_currency: Mapped[str] = mapped_column(String(3), default="USD")
consultation_document: Mapped[dict | None] = mapped_column(FlexibleJSON)
# The MSO's structured written outcome (JSONB, not a separate table)

# MSO doctor-facing
GET  /api/v1/mso/consultations
  → List consultations assigned to this MSO doctor
  → Requires: @require_permission("consultation:read")

GET  /api/v1/mso/consultations/{consultation_id}
  → Case data (MSO redaction policy — clinical only, no PII, no financial)

POST /api/v1/mso/consultations/{consultation_id}/document
  Body: { sections: { clinical_summary, procedure_assessment, ... }, recommendation }
  → Submit consultation document
  → Transitions consultation status → completed

# Patient/coordinator-facing
POST /api/v1/cases/{case_id}/mso/request
  Body: { consultation_format, doctor_id }
  → Create consultation + transition case to mso_offered
  → Requires: case owner or coordinator

GET  /api/v1/cases/{case_id}/mso/status
  → Current MSO consultation status + doctor info + document (if complete)

# Stored as consultation.consultation_document (FlexibleJSON):
{
    "clinical_summary": "MSO's interpretation...",
    "procedure_assessment": "Procedure is appropriate for...",
    "risk_mitigation": "Specific recommendations...",
    "provider_assessment": "Selected provider is suitable...",
    "alternative_options": "None recommended",
    "patient_facing_summary": "Plain language version...",
    "recommendation": "proceed_with_plan",  # proceed_with_plan | consider_alternatives | request_more_info
    "submitted_at": "2026-04-19T..."
}

class VideoRoomService:
    """Stub for Daily.co/100ms room management.

    Phase 2: Create room URL, return to frontend.
    Phase 3: Embedded SDK, recording, participant tracking.
    """

    async def create_room(
        self, consultation_id: str, duration_minutes: int = 30
    ) -> str:
        """Create a video room and return the join URL.

        Phase 2: returns a mock URL or Daily.co API call (gated by flag).
        """

    async def close_room(self, room_id: str) -> None:
        """Close a video room after consultation ends."""

Week 1: Risk assessment
  Day 1-2: RiskScoringService + case model extension + migration
  Day 3-4: Risk review API (queue, assessment, decision)
  Day 5:   Feature flags + integration tests

Week 2: Reviewer dashboard + MSO foundation
  Day 1-2: Coordinator app risk review pages
  Day 3:   MSO doctor model extensions + migration
  Day 4-5: MSO matching service

Week 3: MSO consultation flow
  Day 1-2: Consultation booking service + endpoints
  Day 3-4: Consultation document model + MSO portal
  Day 5:   Video room stub + feature flags

Week 4: Integration + polish
  Day 1-2: End-to-end integration testing
  Day 3:   Coordinator dashboard MSO section
  Day 4-5: Design review + polish

coordinator:
  permissions:
    # ... existing ...
    - case:risk_review:read    # View risk queue
    - case:risk_review:decide  # Approve/reject/request more info