fix(bot): close dialog engine gaps (C-SSRS repeat, scale intros, button payloads, step timeout)

- No same-day C-SSRS repeat: a risk screen completed within cssrs_cooldown_hours is not queued again
  after PHQ-9 item 9 >= 1 (scale crisis path, rule schedule_scale, scale_due); crisis script and alerts stay.
- Scale intro registry: register_scale_intro(scale_id, text_or_fn); start_scale(intro=None) uses it.
  veteran_daily migrated from per-scale on_enter hooks.
- Prefix triggers "button:<prefix>:*" with trigger_payload(event); medication buttons carry the dose
  (med:taken:<dose_id>, ...), med_pending dropped; payload-less legacy buttons answer as stale.
- Abandoned-step timeout: settings.abandon_after_hours (default 6 h) returns abandonable states
  (daily.*, meds.*, scale.* not in onboarding) to idle before the next event; never in crisis or with intake on.

Refs #56

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/dialog/scale_session.py

@@ -15,13 +15,22 @@ Algorithm per item:
   4. Every answer is stored immediately (`item_responses`), so a session interrupted by a crisis keeps its data.
   5. After each answer the scale is scored deterministically (`app.core.scales.score`). If a crisis flag from
      `config/texts/dialog.yaml: settings.crisis_flags` fires (PHQ-9 item 9 ≥ 1, C-SSRS high) the session ends at
-     once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS). A positive
-     item 9 is acted upon even in a half-finished questionnaire.
+     once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS, or a C-SSRS was
+     completed less than `settings.cssrs_cooldown_hours` ago — the patient is not asked the same suicide screen
+     twice in a day; the crisis script and the alert still happen). A positive item 9 is acted upon even in a
+     half-finished questionnaire.
   6. On completion: `scale_results` row (total, severity, item 9, scheduled flag, flags) → the engine commits
      and calls `MonitoringService.after_data_change()` (rules decide alerts — ADR-007) → back to `return_state`.
 The patient never sees a score or a severity label: that would read as a diagnosis.
+
+Intros (#56): `start_scale(intro=None)` looks the intro up in a registry filled by flows
+(`register_scale_intro("phq9", fn_or_text)`), so scales started by the engine (`scale_due`, rule actions) get
+the flow's friendly intro without the flow hooking `scale.<id>` itself.
 """
 
+from collections.abc import Callable
+from dataclasses import replace
+from datetime import timedelta
 from typing import Any
 
 from app.core.privacy import scrub
@@ -29,10 +38,28 @@ from app.core.scales import Item, ScaleDef, ScaleResult, load_scale, score
 from app.db import repo
 from app.dialog.contracts import Button, Event, Reply
 from app.dialog.crisis import crisis_open, open_crisis
-from app.dialog.fsm import Ctx, Step, on_enter, state
+from app.dialog.fsm import RESTING_STATES, Ctx, Step, abandonable, is_abandonable, on_enter, state
 
 PREFIX = "scale"
 
+# (ctx, scale_id, scheduled, return_state) → intro text or None (no intro in this situation).
+IntroFn = Callable[[Ctx, str, bool, str], str | None]
+_intros: dict[str, str | IntroFn] = {}
+
+
+def register_scale_intro(scale_id: str, intro: str | IntroFn, *, replace: bool = False) -> None:
+    """Register the intro shown before `scale_id` when a session is started without an explicit `intro`."""
+    if scale_id in _intros and not replace:
+        raise ValueError(f"intro of scale '{scale_id}' already registered (use replace=True to override)")
+    _intros[scale_id] = intro
+
+
+def scale_intro(ctx: Ctx, scale_id: str, *, scheduled: bool, return_state: str) -> str | None:
+    intro = _intros.get(scale_id)
+    if intro is None or isinstance(intro, str):
+        return intro or None
+    return intro(ctx, scale_id, scheduled, return_state) or None
+
 
 def scale_state(scale_id: str) -> str:
     return f"{PREFIX}.{scale_id}"
@@ -56,9 +83,11 @@ async def start_scale(
     """Begin a session of `scale_id`; returns a Step moving the FSM to `scale.<id>`.
 
     `scheduled=False` marks an unscheduled administration (triggered by a rule or a safety signal).
-    `intro` — optional extra text shown before the scale instruction.
+    `intro` — optional extra text shown before the scale instruction; None → the registered intro, if any.
     """
     scale = load_scale(scale_id)  # fail fast on an unknown id
+    if intro is None:
+        intro = scale_intro(ctx, scale_id, scheduled=scheduled, return_state=return_state)
     sessions = dict(ctx.data.get("scales") or {})
     sessions[scale_id] = {
         "idx": 0,
@@ -98,6 +127,28 @@ def _session(ctx: Ctx) -> tuple[ScaleDef, dict[str, Any]] | None:
     return load_scale(scale_id), session
 
 
+def _abandon_allowed(ctx: Ctx) -> bool:
+    """A silent scale session may be abandoned only if what it returns to may be (idle, a check-in step) —
+    never an onboarding chain, which would otherwise be skipped."""
+    found = _session(ctx)
+    if found is None:
+        return True
+    back = str(found[1].get("return_state") or "idle")
+    if back in RESTING_STATES:
+        return True
+    return back != ctx.state and is_abandonable(replace(ctx, state=back))
+
+
+abandonable(PREFIX, _abandon_allowed)
+
+
+async def recent_risk_screen(ctx: Ctx, screen: str) -> bool:
+    """A complete risk screen (C-SSRS) less than `cssrs_cooldown_hours` ago (scale_results, any trigger)."""
+    since = ctx.now - timedelta(hours=ctx.texts.settings.cssrs_cooldown_hours)
+    results = await repo.list_scale_results(ctx.session, ctx.token, screen)
+    return any(r.complete and r.ts >= since for r in results)
+
+
 @on_enter(PREFIX)
 async def _enter(ctx: Ctx) -> Step:
     found = _session(ctx)
@@ -231,6 +282,7 @@ async def _finish(
     else:
         crisis = open_crisis(ctx, categories=[], source=f"scale:{scale.id}")
     step = done.then(crisis)
-    if scale.id != risk_screen:
+    # Same-day repeat guard: a C-SSRS completed within the cooldown already answered the screening question.
+    if scale.id != risk_screen and not await recent_risk_screen(ctx, risk_screen):
         step.data["pending_scales"] = add_pending(ctx.data, risk_screen, scheduled=False)
     return step