feat(questionnaire): integrate revised intake and follow-up monitoring

Add gated TSQ, Russian C-SSRS Recent, durable responses and notifications, diaries, schedules and audited clinical review. Preserve manual assessment when the Russian follow-up source is unavailable.

Refs #52

app/dialog/engine.py

@@ -22,15 +22,18 @@ Pipeline for one incoming event (SPEC §3, guardrails #1, #2, #3):
 Logs carry only the token, states and event codes (guardrail #8).
 """
 
+import asyncio
 import logging
 from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Any
+from weakref import WeakValueDictionary
 
 from sqlalchemy.exc import InterfaceError, OperationalError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core import clock, safety
+from app.core.intake_runtime import intake_enabled
 from app.core.privacy import scrub
 from app.core.safety import SafetyResult, crisis_reply
 from app.core.types import RiskLevel, ScrubbedText
@@ -45,12 +48,14 @@ from app.dialog.flows.core import close_crisis, revoke
 from app.dialog.fsm import NO_CONSENT_STATES, RESTING_STATES, Ctx, Step, apply_data, enter_hook_for, handler_for
 from app.dialog.texts import DialogTexts, load_dialog_texts
 from app.llm.gateway import LLMGateway, get_gateway
+from app.services import questionnaire as intake_service
 from app.services.contracts import MonitoringService, get_monitoring
 
 log = logging.getLogger(__name__)
 
 # Enter hooks may chain (scale done → next onboarding state → next scale); a cycle must not hang a request.
 MAX_ENTER_CHAIN = 8
+_patient_locks = WeakValueDictionary()
 # DB unreachable: answer safely instead of failing the channel (connectivity errors only — bugs must surface).
 _DB_DOWN = (OSError, OperationalError, InterfaceError)
 
@@ -76,8 +81,13 @@ class Engine:
     async def handle(self, token: str, event: Event) -> list[Reply]:
         texts = load_dialog_texts()
         try:
-            async with get_clinical_session() as session:
-                return await self._handle(token, event, session, texts)
+            lock = _patient_locks.setdefault(token, asyncio.Lock())
+            async with lock:
+                async with get_clinical_session() as session:
+                    await intake_service.lock_patient(session, token)
+                    replies = await self._handle(token, event, session, texts)
+                await intake_service.dispatch_notices(token)
+                return replies
         except _DB_DOWN as exc:
             log.error("dialog db unavailable token=%s event=%s error=%s", token, event.type, type(exc).__name__)
             return _degraded(event, texts)
@@ -117,6 +127,13 @@ class Engine:
                     session, token=token, direction="out", text=scrub(reply.text), ts=ctx.now, fsm_state=ctx.state
                 )
         await self._save_state(ctx)
+        task_id = event.params.get("questionnaire_task")
+        if event.type == "system" and task_id and run.replies:
+            from app.db.intake import QuestionnaireTask
+
+            task = await session.get(QuestionnaireTask, task_id)
+            if task and task.token == token:
+                task.delivery = [r.model_dump() for r in run.replies]
         log.info(
             "dialog token=%s event=%s code=%s state=%s->%s replies=%d",
             token,
@@ -143,6 +160,12 @@ class Engine:
         if event.type == "button":
             if consented and event.button == "consent:revoke":
                 return await revoke(ctx)
+            if consented and event.button == "intake:help":
+                from app.db.intake import QuestionnaireRecord
+                from app.dialog.flows.intake import urgent
+
+                row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
+                return await urgent(ctx, row)
             if consented:
                 await self._store_in(ctx, scrub(f"[button] {event.button}"))  # button data comes from the client
             return await self._dispatch(ctx, event)
@@ -153,6 +176,14 @@ class Engine:
             if not consented:
                 return Step()  # nothing is pushed to people without consent
             if event.system == "scale_due" and event.params.get("scale"):
+                if intake_enabled() and ctx.data.get("adult") is True:
+                    scale = str(event.params["scale"])
+                    kind = "cssrs_since_contact" if scale in ("cssrs_screen", "cssrs") else scale
+                    if kind in ("phq9", "pcl5", "rus_audit_s", "cssrs_since_contact"):
+                        await intake_service.queue_task(
+                            ctx.session, ctx.token, kind, f"external:{kind}:{ctx.now.date()}", ctx.now
+                        )
+                    return Step()
                 scheduled = bool(event.params.get("scheduled", True))
                 pending = scale_session.add_pending(ctx.data, str(event.params["scale"]), scheduled)
                 return Step(data={"pending_scales": pending})  # starts now if resting, else after the current step
@@ -181,11 +212,29 @@ class Engine:
         if result.level == RiskLevel.ACUTE:
             handled = "crisis_repeat" if crisis_open(ctx.data) else "crisis_script"
             log.warning("safety acute token=%s rules=%s handled=%s", ctx.token, result.matched_rules, handled)
-            step = open_crisis(ctx, result.categories, source=f"safety:{result.source}")
+            if ctx.data.get("questionnaire_record") or (intake_enabled() and ctx.data.get("adult") is True):
+                from app.db.intake import QuestionnaireRecord
+                from app.dialog.flows.intake import urgent
+
+                row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
+                step = await urgent(ctx, row, code="safety_acute", categories=result.categories)
+            else:
+                step = open_crisis(ctx, result.categories, source=f"safety:{result.source}")
             step.data["safety_events"] = self._safety_events(ctx, result, handled)
             step.notify_monitoring = True  # lane B escalates to the doctor on call
             return step
 
+        if result.level == RiskLevel.POSSIBLE and (
+            ctx.data.get("questionnaire_record") or (intake_enabled() and ctx.data.get("adult") is True)
+        ):
+            from app.db.intake import QuestionnaireRecord
+            from app.dialog.flows.intake import manual_review
+
+            row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
+            if row is None:
+                row = await intake_service.create_record(ctx, "cssrs_since_contact")
+            return await manual_review(ctx, row, "cssrs_since_contact_required")
+
         if result.level == RiskLevel.POSSIBLE:
             screen = ctx.texts.settings.risk_screen_scale
             if self._can_screen(ctx, screen):
@@ -252,6 +301,8 @@ class Engine:
 
     async def _start_pending(self, ctx: Ctx) -> Step:
         """Start a queued scale when nothing else is being asked. In crisis only the risk screen may run."""
+        if intake_enabled():
+            return Step()
         pending = list(ctx.data.get("pending_scales") or [])
         if not pending or ctx.state not in RESTING_STATES:
             return Step()