Align endpoint contract with challenge brief

README.md

@@ -10,24 +10,29 @@ pinned: false
 
 Submission bot for the magicpin Vera AI Challenge. It exposes the required HTTP API, stores judge-pushed context in memory, and composes grounded merchant/customer actions from the JSON context it receives.
 
-The live bot does not require a paid API key. If `OPENAI_API_KEY` is present, it can polish copy after deterministic planning, but the deterministic engine remains the source of truth and invalid LLM output is discarded. OpenRouter can be used only for offline G-Eval-style calibration.
+The submitted live bot is deterministic and does not require a paid API key or any runtime LLM call. OpenRouter/LLM tooling is used only for offline calibration and local judging experiments, never as a dependency for `/v1/tick`.
 
 ## How It Works
 
 1. The judge pushes category, merchant, customer, and trigger JSON into `POST /v1/context`.
 2. `POST /v1/tick` ranks active triggers by expected rubric score and returns up to 20 message actions.
 3. `POST /v1/reply` handles auto-replies, commitment, off-topic replies, STOP/hostility, and ended conversations.
-4. `POST /v1/teardown` clears in-memory state for clean reruns.
+4. `POST /v1/teardown` is an extra helper for clean local/HF reruns; it is not required by the challenge judge.
 
-Required endpoints:
+Challenge-required endpoints:
 
 - `GET /v1/healthz`
 - `GET /v1/metadata`
 - `POST /v1/context`
 - `POST /v1/tick`
 - `POST /v1/reply`
+
+Additional debug/test endpoint:
+
 - `POST /v1/teardown`
 
+Compatibility aliases without the `/v1` prefix are also exposed (`/healthz`, `/metadata`, `/context`, `/tick`, `/reply`) because the public challenge page mentions both shorthand endpoint names and the `/v1/...` submission contract. The submitted base URL remains the same.
+
 ## Scoring Strategy
 
 The composer is a rubric-optimized decision engine:
@@ -66,7 +71,7 @@ pytest -q
 python -m compileall app bot.py scripts tests
 python scripts/generate_submission.py
 python scripts/lint_submission.py
-python scripts/score_proxy.py 34
+python scripts/score_proxy.py 43
 python scripts/geval_calibrate.py
 ```
 
@@ -83,14 +88,14 @@ python scripts/run_judge_with_trigger_log.py
 
 It prints the lowest-scoring trigger IDs and can write JSON when `JUDGE_TRIGGER_LOG_JSON` is set.
 
-## Optional LLM Copy Polish
+## Optional Local LLM Copy Polish
 
-Set these only if you want controlled copy improvement:
+The submitted HF deployment does not use runtime LLM polish. Set these only for local experiments:
 
 - `OPENAI_API_KEY`
 - `OPENAI_MODEL`, default `gpt-4o-mini`
 
-The model receives only the deterministic plan and evidence. It must return structured JSON with `body`, `cta`, `send_as`, `suppression_key`, and `rationale`. The bot falls back to deterministic copy if the call fails, times out, changes protected fields, or invents numbers.
+The model receives only the deterministic plan and evidence. It must return structured JSON with `body`, `cta`, `send_as`, `suppression_key`, and `rationale`. The bot falls back to deterministic copy if the call fails, times out, changes protected fields, or invents numbers. Do not enable this in the submitted Space; deterministic runtime is the reliability tradeoff.
 
 ## Optional OpenRouter Calibration
 

app/composer.py

@@ -1,668 +1,668 @@
-from __future__ import annotations
-
-from datetime import datetime
-from typing import Any
-import hashlib
-import re
-
-from .decision_engine import compose_scored
-
-
-Context = dict[str, Any]
-
-
-CTA_NONE = "none"
-CTA_OPEN = "open_ended"
-CTA_YES_NO = "binary_yes_no"
-CTA_CONFIRM = "binary_confirm_cancel"
-CTA_SLOTS = "multi_choice_slot"
-
-
-def compose(category: Context, merchant: Context, trigger: Context, customer: Context | None = None) -> Context:
-    """Compose a deterministic, context-grounded Vera message."""
-    scored = compose_scored(category, merchant, trigger, customer)
-    if scored:
-        return validate_message(scored)
-    if customer:
-        return _compose_customer(category, merchant, trigger, customer)
-    return _compose_merchant(category, merchant, trigger)
-
-
-def reply_to_message(
-    message: str,
-    conversation: Context | None = None,
-    merchant: Context | None = None,
-    trigger: Context | None = None,
-) -> Context:
-    """Respond to a merchant/customer reply without using an LLM."""
-    text = (message or "").strip()
-    lower = text.lower()
-    conversation = conversation or {}
-    auto_count = int(conversation.get("auto_reply_count", 0))
-
-    if is_stop_or_hostile(lower):
-        return {
-            "action": "end",
-            "rationale": "The sender explicitly rejected further messages or used hostile stop language; ending without another nudge.",
-        }
-
-    if is_auto_reply(text):
-        if auto_count >= 3:
-            return {
-                "action": "end",
-                "rationale": "Repeated canned auto-reply detected three times; closing the conversation to avoid wasting turns.",
-            }
-        if auto_count == 2:
-            return {
-                "action": "wait",
-                "wait_seconds": 86400,
-                "rationale": "Same auto-reply repeated; owner is likely unavailable, so Vera waits 24 hours.",
-            }
-        return {
-            "action": "wait",
-            "wait_seconds": 14400,
-            "rationale": "Canned WhatsApp Business auto-reply detected; backing off 4 hours for a real owner reply.",
-        }
-
-    if is_commitment(lower):
-        scope = _action_scope(merchant or {}, trigger or {})
-        return {
-            "action": "send",
-            "body": f"Done. I am preparing {scope} now from the details already shared. I will keep it to one ready-to-send draft/action and avoid adding anything not in your current context.",
-            "cta": CTA_NONE,
-            "rationale": "The sender committed; switching directly to action mode without another qualification or confirmation loop.",
-        }
-
-    if is_offtopic(lower):
-        return {
-            "action": "send",
-            "body": "That is outside what I can help with directly. Coming back to this Vera task: should I prepare the draft/action from the details above?",
-            "cta": CTA_YES_NO,
-            "rationale": "Politely declines an off-topic request and returns to the active merchant-growth task.",
-        }
-
-    if is_delay(lower):
-        return {
-            "action": "wait",
-            "wait_seconds": 1800,
-            "rationale": "The sender asked for time or signaled they are busy; wait 30 minutes.",
-        }
-
-    if "?" in text or any(w in lower for w in ["what", "how", "price", "cost", "details", "send"]):
-        return {
-            "action": "send",
-            "body": "Yes. I can keep it simple: I will draft one ready-to-send version using only your current offer, locality, and the trigger we discussed. Reply YES and I will prepare it.",
-            "cta": CTA_YES_NO,
-            "rationale": "The sender is engaged and asking for details; answer briefly and request one low-friction confirmation.",
-        }
-
-    return {
-        "action": "send",
-        "body": "Got it. Should I prepare the next draft/action for this now?",
-        "cta": CTA_YES_NO,
-        "rationale": "Acknowledges an ambiguous but non-negative reply and asks for one clear next step.",
-    }
-
-
-def is_auto_reply(message: str) -> bool:
-    lower = (message or "").lower().strip()
-    patterns = [
-        "thank you for contacting",
-        "thanks for contacting",
-        "our team will respond",
-        "we will respond shortly",
-        "we are currently unavailable",
-        "business hours",
-        "automated assistant",
-        "auto-reply",
-        "away message",
-        "will get back to you",
-    ]
-    return any(p in lower for p in patterns)
-
-
-def is_stop_or_hostile(lower: str) -> bool:
-    if re.search(r"\b(stop|unsubscribe|remove me|opt out)\b", lower):
-        return True
-    hard_stops = [
-        "stop messaging",
-        "stop sending",
-        "unsubscribe",
-        "not interested",
-        "dont message",
-        "don't message",
-        "useless spam",
-        "spam",
-        "leave me",
-        "bothering me",
-    ]
-    return any(p in lower for p in hard_stops)
-
-
-def is_commitment(lower: str) -> bool:
-    commitments = [
-        r"\byes\b",
-        r"\bok\b",
-        r"\bokay\b",
-        r"\bgo ahead\b",
-        r"\blets do it\b",
-        r"\blet's do it\b",
-        r"\bconfirm\b",
-        r"\bproceed\b",
-        r"\bsend it\b",
-        r"\bdo it\b",
-        r"\bstart\b",
-        r"\bi want to join\b",
-        r"\bmujhe magicpin\b",
-        r"\bchalo\b",
-    ]
-    return any(re.search(pattern, lower) for pattern in commitments)
-
-
-def is_delay(lower: str) -> bool:
-    return any(p in lower for p in ["later", "busy", "after some time", "tomorrow", "call later", "not now"])
-
-
-def is_offtopic(lower: str) -> bool:
-    return any(p in lower for p in ["gst", "tax", "income tax", "hiring", "salary", "loan", "rent agreement", "website design", "cricket score", "weather", "movie ticket"])
-
-
-def validate_message(message: Context) -> Context:
-    """Normalize shape and remove risky leftovers."""
-    body = _clean_multiline(str(message.get("body", "")))
-    if not body:
-        body = "Quick Vera update: I found one relevant action from your current context. Want me to prepare it?"
-    cta = message.get("cta") or CTA_OPEN
-    if cta not in {CTA_NONE, CTA_OPEN, CTA_YES_NO, CTA_CONFIRM, CTA_SLOTS, "binary_yes_stop"}:
-        cta = CTA_OPEN
-    body = _final_scrub(body)
-    message["body"] = body[:1800]
-    message["cta"] = cta
-    message.setdefault("send_as", "vera")
-    message.setdefault("suppression_key", "")
-    message.setdefault("rationale", "Composed deterministically from category, merchant, trigger, and optional customer context.")
-    return message
-
-
-def make_conversation_id(merchant_id: str, trigger_id: str, customer_id: str | None = None) -> str:
-    base = f"{merchant_id}:{trigger_id}:{customer_id or ''}"
-    digest = hashlib.sha1(base.encode("utf-8")).hexdigest()[:8]
-    merchant_short = _short_id(merchant_id)
-    trigger_short = _short_id(trigger_id)
-    if customer_id:
-        return f"conv_{merchant_short}_{_short_id(customer_id)}_{digest}"
-    return f"conv_{merchant_short}_{trigger_short}_{digest}"
-
-
-def _compose_merchant(category: Context, merchant: Context, trigger: Context) -> Context:
-    kind = trigger.get("kind", "generic")
-    cat = merchant.get("category_slug") or category.get("slug", "merchant")
-    identity = merchant.get("identity", {})
-    name = _merchant_name(merchant)
-    first = _owner_or_name(merchant)
-    payload = trigger.get("payload", {})
-    offer = _best_offer(merchant, category)
-    perf = merchant.get("performance", {})
-    agg = merchant.get("customer_aggregate", {})
-    locality = identity.get("locality") or identity.get("city") or "your area"
-
-    if _is_placeholder_payload(payload):
-        return _generic_merchant(category, merchant, trigger)
-
-    if kind in {"research_digest", "regulation_change", "cde_opportunity"}:
-        item = _digest_item(category, payload.get("top_item_id") or payload.get("digest_item_id"))
-        title = item.get("title") or payload.get("metric_or_topic") or "this week's category update"
-        source = item.get("source") or "your category digest"
-        number = _first_fact(item, agg)
-        if kind == "regulation_change":
-            deadline = payload.get("deadline_iso") or trigger.get("expires_at", "")[:10]
-            body = f"{name}, compliance note: {title}. Deadline: {deadline}. {number} Source: {source}. Want me to draft the 5-point SOP/checklist for your clinic?"
-            cta = CTA_YES_NO
-        elif kind == "cde_opportunity":
-            credits = payload.get("credits") or item.get("credits")
-            fee = str(payload.get("fee") or item.get("actionable") or "").replace("_", " ")
-            body = f"{name}, IDA/CDE item for you: {title}. Credits: {credits}; fee: {fee}. Source: {source}. Want me to pull the invite and make a 1-line calendar reminder?"
-            cta = CTA_YES_NO
-        else:
-            cohort = _cohort_phrase(agg)
-            body = f"{name}, {source} has one item relevant to {cohort}: {title}. {number} Want me to pull the 2-min summary and draft a patient WhatsApp?"
-            cta = CTA_OPEN
-        return _msg(body, cta, "vera", trigger, f"{kind} uses digest source plus merchant cohort/performance context.")
-
-    if kind in {"perf_dip", "seasonal_perf_dip"}:
-        if kind == "perf_dip" and not _has_all(payload, "metric", "delta_pct", "window"):
-            return _generic_merchant(category, merchant, trigger)
-        metric = payload.get("metric", "performance")
-        delta = _pct(payload.get("delta_pct") or perf.get("delta_7d", {}).get(f"{metric}_pct"))
-        baseline = payload.get("vs_baseline")
-        seasonal = payload.get("is_expected_seasonal")
-        if seasonal:
-            body = f"{first}, your {metric} is down {delta} this {payload.get('window', 'week')}, but this matches the {payload.get('season_note', 'seasonal dip')} pattern. Do not over-spend ads now; focus on your {_member_count(agg)}. Want me to draft a retention nudge?"
-        else:
-            body = f"{first}, {metric} dropped {delta} in {payload.get('window', '7d')}; baseline was {baseline} and current calls are {perf.get('calls', 'lower than usual')}. {offer} is the quickest concrete hook. Want me to draft a recovery WhatsApp/GBP post?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Performance dip trigger; recommends one low-effort recovery action.")
-
-    if kind == "perf_spike":
-        if not _has_all(payload, "metric", "delta_pct", "window"):
-            return _generic_merchant(category, merchant, trigger)
-        metric = payload.get("metric", "calls")
-        delta = _pct(payload.get("delta_pct"))
-        driver = str(payload.get("likely_driver") or "recent profile activity").replace("_", " ")
-        body = f"{first}, {metric} is up {delta} in {payload.get('window', '7d')} vs baseline {payload.get('vs_baseline', 'normal')}. Likely driver: {driver}. Want me to turn this into a repeatable post for this week?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Performance spike trigger; converts a winning signal into a repeatable action.")
-
-    if kind in {"active_planning_intent"}:
-        topic = str(payload.get("intent_topic", "growth plan")).replace("_", " ")
-        last = payload.get("merchant_last_message", "")
-        if "thali" in topic:
-            body = f"{first}, based on your '{last}' message, here is a starter corporate thali structure: 10 thalis at the current {offer}, 25+ with free delivery, 50+ with a filter-coffee add-on. Want me to draft the 3-line outreach note?"
-        elif "kids_yoga" in topic:
-            body = f"{first}, for kids yoga summer camp, keep it simple: age 7-12, 4 weeks, 3 sessions/week, Saturday trial, and use your current {offer}. Want me to draft the GBP post plus parent WhatsApp?"
-        else:
-            body = f"{first}, you asked about {topic}. I can turn it into one concrete offer using {offer}, your {locality} locality, and your current profile. Want the ready draft?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Merchant already showed planning intent; moves directly to a concrete draft.")
-
-    if kind == "festival_upcoming":
-        if not _has_all(payload, "festival", "days_until"):
-            return _generic_merchant(category, merchant, trigger)
-        festival = payload.get("festival", "festival")
-        days = payload.get("days_until")
-        body = f"{first}, {festival} is {days} days away and {cat} is marked relevant for this beat. Your active hook is {offer}. Want me to prepare a festival post now and hold it for approval?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Festival trigger with category relevance and existing offer.")
-
-    if kind == "ipl_match_today":
-        if not _has_all(payload, "match", "venue", "match_time_iso"):
-            return _generic_merchant(category, merchant, trigger)
-        match = payload.get("match", "today's match")
-        venue = payload.get("venue", "the stadium")
-        time = _time_from_iso(payload.get("match_time_iso")) or "tonight"
-        weeknight = payload.get("is_weeknight")
-        advice = "push a delivery-only offer" if not weeknight else "run a quick pre-match dine-in/post"
-        body = f"{first}, {match} at {venue} starts {time}. Since this is {'not ' if not weeknight else ''}a weeknight match, {advice} using your active {offer}. Want me to draft the banner text and Insta story?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "IPL trigger interpreted with day context and current restaurant offer.")
-
-    if kind == "review_theme_emerged":
-        if not _has_all(payload, "theme", "occurrences_30d"):
-            return _generic_merchant(category, merchant, trigger)
-        theme = str(payload.get("theme", "review theme")).replace("_", " ")
-        count = payload.get("occurrences_30d")
-        quote = payload.get("common_quote")
-        body = f"{first}, {count} reviews in 30d now mention {theme}; one customer said '{quote}'. Want me to draft a short public reply pattern plus an ops checklist?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Review theme trigger turns repeated feedback into reply and ops action.")
-
-    if kind == "milestone_reached":
-        if not _has_all(payload, "metric", "value_now", "milestone_value"):
-            return _generic_merchant(category, merchant, trigger)
-        value = payload.get("value_now")
-        target = payload.get("milestone_value")
-        metric = str(payload.get("metric", "milestone")).replace("_", " ")
-        body = f"{first}, you are at {value} {metric}; {target} is close. Want me to draft a polite review-request WhatsApp for recent happy customers?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Milestone trigger nudges a review/request action tied to current count.")
-
-    if kind == "renewal_due":
-        if not _has_any(payload, "days_remaining", "renewal_amount") and not merchant.get("subscription", {}).get("days_remaining"):
-            return _generic_merchant(category, merchant, trigger)
-        days = payload.get("days_remaining") or merchant.get("subscription", {}).get("days_remaining")
-        amount = payload.get("renewal_amount")
-        body = f"{first}, your {payload.get('plan', 'plan')} renewal is due in {days} days for Rs {amount}. Calls are down {_pct(merchant.get('performance', {}).get('delta_7d', {}).get('calls_pct'))}; before renewal, want me to show the 3 fixes likely to recover calls?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Renewal trigger pairs deadline with current performance risk.")
-
-    if kind in {"winback_eligible", "dormant_with_vera"}:
-        if not _has_any(payload, "days_since_expiry", "days_since_last_merchant_message", "lapsed_customers_added_since_expiry"):
-            return _generic_merchant(category, merchant, trigger)
-        days = payload.get("days_since_expiry") or payload.get("days_since_last_merchant_message") or merchant.get("subscription", {}).get("days_since_expiry")
-        lapsed = payload.get("lapsed_customers_added_since_expiry") or agg.get("lapsed_90d_plus") or agg.get("lapsed_180d_plus")
-        body = f"{first}, it has been {days} days since the last active Vera/subscription moment. You now have {lapsed} lapsed customers/signals to recover. Want me to draft one winback message using {offer}?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Dormancy/winback trigger; restarts with one concrete recovery action.")
-
-    if kind == "supply_alert":
-        if not _has_all(payload, "molecule", "affected_batches"):
-            return _generic_merchant(category, merchant, trigger)
-        batches = ", ".join(payload.get("affected_batches", [])[:3])
-        molecule = payload.get("molecule", "medicine")
-        chronic = agg.get("chronic_rx_count") or agg.get("total_unique_ytd") or "repeat"
-        body = f"{first}, urgent stock alert: {molecule} batches {batches} from {payload.get('manufacturer', 'the manufacturer')}. You have {chronic} chronic-Rx customers in context. Want me to draft the replacement WhatsApp and counter checklist?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Supply alert trigger uses molecule, batch numbers, and pharmacy repeat-customer context.")
-
-    if kind == "category_seasonal":
-        if not payload.get("trends"):
-            return _generic_merchant(category, merchant, trigger)
-        trends = ", ".join(str(t).replace("_", " ") for t in payload.get("trends", [])[:4])
-        body = f"{first}, summer demand shift is visible: {trends}. Since shelf action is recommended, want me to draft a 10-item reorder/checklist plus WhatsApp note?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Seasonal pharmacy/category trigger maps trends to stock action.")
-
-    if kind == "gbp_unverified":
-        if not _has_all(payload, "verification_path", "estimated_uplift_pct"):
-            return _generic_merchant(category, merchant, trigger)
-        uplift = _pct(payload.get("estimated_uplift_pct"))
-        body = f"{first}, your GBP is still unverified; the available path is {payload.get('verification_path', 'verification')}. Verified profiles can unlock about {uplift} more visibility in this context. Want me to walk you through the 3-step verification?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "GBP verification trigger uses exact path and estimated uplift.")
-
-    if kind == "competitor_opened":
-        if not _has_all(payload, "competitor_name", "distance_km", "opened_date", "their_offer"):
-            return _generic_merchant(category, merchant, trigger)
-        body = f"{name}, new competitor signal: {payload.get('competitor_name')} opened {payload.get('distance_km')} km away on {payload.get('opened_date')} with {payload.get('their_offer')}. Your current hook is {offer}. Want me to draft a sharper local post?"
-        return _msg(body, CTA_YES_NO, "vera", trigger, "Competitor trigger uses named competitor, distance, date, and offer comparison.")
-
-    if kind == "curious_ask_due":
-        body = f"{first}, quick check: what service has been most asked-for this week at {identity.get('name', 'your business')}? I will turn your answer into a Google post and a 4-line WhatsApp reply. Takes 5 min."
-        return _msg(body, CTA_OPEN, "vera", trigger, "Curious-ask cadence asks the merchant for one useful signal.")
-
-    return _generic_merchant(category, merchant, trigger)
-
-
-def _compose_customer(category: Context, merchant: Context, trigger: Context, customer: Context) -> Context:
-    if not _has_consent(customer, trigger):
-        return _msg(
-            f"{_owner_or_name(merchant)}, {_customer_name(customer)} has a {str(trigger.get('kind', 'customer')).replace('_', ' ')} signal, but the saved opt-in scope does not clearly cover this outreach. Want me to prepare a consent-safe approval note first?",
-            CTA_YES_NO,
-            "vera",
-            trigger,
-            "Customer trigger routed to merchant because consent/preference does not clearly permit direct outreach.",
-        )
-
-    kind = trigger.get("kind", "customer_message")
-    payload = trigger.get("payload", {})
-    merchant_name = merchant.get("identity", {}).get("name", "your merchant")
-    owner = merchant.get("identity", {}).get("owner_first_name") or merchant_name
-    customer_name = _customer_name(customer)
-    offer = _best_offer(merchant, category)
-    lang = str(customer.get("identity", {}).get("language_pref", "english")).lower()
-
-    if _is_placeholder_payload(payload):
-        return _generic_customer(category, merchant, trigger, customer)
-
-    if kind == "recall_due":
-        if not _has_any(payload, "service_due", "due_date", "available_slots"):
-            return _generic_customer(category, merchant, trigger, customer)
-        slots = payload.get("available_slots", [])
-        slot_text = _slot_text(slots)
-        months = _months_between(payload.get("last_service_date"), payload.get("due_date")) or "6-month"
-        if "hi" in lang:
-            body = f"Hi {customer_name}, {merchant_name} here. It has been about {months} since your last visit; your {payload.get('service_due', 'recall').replace('_', ' ')} is due. Apke liye slots: {slot_text}. {offer}. Reply 1/2 for a slot, or suggest a time."
-        else:
-            body = f"Hi {customer_name}, {merchant_name} here. It has been about {months} since your last visit; your {payload.get('service_due', 'recall').replace('_', ' ')} is due. Slots available: {slot_text}. {offer}. Reply 1/2 for a slot, or suggest a time."
-        return _msg(body, CTA_SLOTS, "merchant_on_behalf", trigger, "Customer recall uses due date, slot options, offer, and language preference.")
-
-    if kind in {"customer_lapsed_hard", "customer_lapsed_soft"}:
-        days = payload.get("days_since_last_visit")
-        focus = str(payload.get("previous_focus") or customer.get("preferences", {}).get("training_focus") or "your goal").replace("_", " ")
-        elapsed = f"{days} days" if days is not None else "a while"
-        body = f"Hi {customer_name}, {owner} from {merchant_name} here. It has been {elapsed} since your last visit; no pressure. We can restart with {offer}, matched to {focus}. Reply YES and I will hold a no-commitment slot."
-        return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Customer lapse trigger uses days since visit, prior focus, and merchant offer.")
-
-    if kind == "appointment_tomorrow":
-        appointment = payload.get("appointment_time") or payload.get("slot_label") or _slot_text(payload.get("available_slots", []))
-        service = str(payload.get("service") or customer.get("relationship", {}).get("services_received", ["appointment"])[-1]).replace("_", " ")
-        body = f"Hi {customer_name}, {merchant_name} here. Reminder for your {service} appointment tomorrow: {appointment}. Reply YES to confirm or tell us if you need to reschedule."
-        return _msg(body, CTA_CONFIRM, "merchant_on_behalf", trigger, "Appointment reminder uses customer relationship and available appointment timing.")
-
-    if kind == "wedding_package_followup":
-        if not _has_any(payload, "wedding_date", "days_to_wedding", "next_step_window_open"):
-            return _generic_customer(category, merchant, trigger, customer)
-        days = payload.get("days_to_wedding")
-        body = f"Hi {customer_name}, {owner} from {merchant_name} here. {days} days to your wedding; this is the right window for {str(payload.get('next_step_window_open', 'skin prep')).replace('_', ' ')}. {offer}. Want me to block your preferred Saturday slot?"
-        return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Bridal follow-up uses wedding date window, relationship history, and offer.")
-
-    if kind == "trial_followup":
-        if not _has_any(payload, "trial_date", "next_session_options"):
-            return _generic_customer(category, merchant, trigger, customer)
-        slots = _slot_text(payload.get("next_session_options", []))
-        body = f"Hi {customer_name}, {owner} from {merchant_name} here. Thanks for trying the class on {payload.get('trial_date')}. Next available option: {slots}. Want me to reserve it?"
-        return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Trial follow-up uses trial date and next available session.")
-
-    if kind == "chronic_refill_due":
-        if not _has_any(payload, "molecule_list", "stock_runs_out_iso"):
-            return _generic_customer(category, merchant, trigger, customer)
-        meds = ", ".join(payload.get("molecule_list", []))
-        address_note = "We have your saved delivery address on file." if payload.get("delivery_address_saved") else "We can confirm the delivery address after your reply."
-        body = f"Namaste {customer_name}, {merchant_name} here. Your monthly medicines ({meds}) are due by {_date_from_iso(payload.get('stock_runs_out_iso'))}. {address_note} Reply CONFIRM and our pharmacist will verify stock and delivery details before preparing it."
-        return _msg(body, CTA_CONFIRM, "merchant_on_behalf", trigger, "Refill reminder uses molecule list, run-out date, delivery status, and pharmacy offer.")
-
-    return _msg(
-        f"Hi {customer_name}, {merchant_name} here. A quick update related to your last visit is ready. {offer}. Reply YES if you want the details.",
-        CTA_YES_NO,
-        "merchant_on_behalf",
-        trigger,
-        "Generic customer-scoped trigger with consent and merchant offer.",
-    )
-
-
-def _generic_merchant(category: Context, merchant: Context, trigger: Context) -> Context:
-    first = _owner_or_name(merchant)
-    kind = str(trigger.get("kind", "update")).replace("_", " ")
-    perf = merchant.get("performance", {})
-    offer = _best_offer(merchant, category)
-    views = perf.get("views")
-    calls = perf.get("calls")
-    identity = merchant.get("identity", {})
-    locality = identity.get("locality") or identity.get("city") or "your locality"
-    fact = f"Current 30d views: {views}; calls: {calls}." if views is not None and calls is not None else f"Locality: {locality}."
-    body = f"{first}, Vera found a {kind} signal for {_merchant_name(merchant)}. {fact} Best hook: {offer}. Want me to draft the next WhatsApp/GBP action?"
-    return _msg(body, CTA_YES_NO, "vera", trigger, "Fallback grounded in trigger kind, performance, and offer.")
-
-
-def _generic_customer(category: Context, merchant: Context, trigger: Context, customer: Context) -> Context:
-    customer_name = _customer_name(customer)
-    merchant_name = merchant.get("identity", {}).get("name", "your merchant")
-    kind = str(trigger.get("kind", "customer update")).replace("_", " ")
-    last_visit = customer.get("relationship", {}).get("last_visit")
-    state = str(customer.get("state", "active")).replace("_", " ")
-    offer = _best_offer(merchant, category)
-    visit_text = f" Last visit: {last_visit}." if last_visit else ""
-    body = f"Hi {customer_name}, {merchant_name} here. Quick {kind} update for you based on your {state} status.{visit_text} {offer}. Reply YES if you want me to hold the next step."
-    return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Customer fallback grounded in relationship state, last visit, and available offer.")
-
-
-def _msg(body: str, cta: str, send_as: str, trigger: Context, rationale: str) -> Context:
-    return validate_message({
-        "body": body,
-        "cta": cta,
-        "send_as": send_as,
-        "suppression_key": trigger.get("suppression_key", trigger.get("id", "")),
-        "rationale": rationale,
-    })
-
-
-def _clean(text: str) -> str:
-    return re.sub(r"\s+", " ", text).strip()
-
-
-def _clean_multiline(text: str) -> str:
-    text = re.sub(r"\.{2,}", ".", text)
-    text = re.sub(r";\s*\.", ".", text)
-    text = re.sub(r"[ \t]+", " ", text)
-    text = re.sub(r" *\n *", "\n", text)
-    text = re.sub(r"\n{3,}", "\n\n", text)
-    return text.strip()
-
-
-def _merchant_name(merchant: Context) -> str:
-    identity = merchant.get("identity", {})
-    name = identity.get("name") or "Merchant"
-    owner = str(identity.get("owner_first_name") or "").strip()
-    if merchant.get("category_slug") == "dentists" and owner and not str(name).lower().startswith("dr."):
-        return owner if owner.lower().startswith("dr") else f"Dr. {owner}"
-    return _dedupe_dr(str(name))
-
-
-def _owner_or_name(merchant: Context) -> str:
-    identity = merchant.get("identity", {})
-    owner = identity.get("owner_first_name")
-    if owner:
-        if merchant.get("category_slug") == "dentists" and not str(owner).lower().startswith("dr"):
-            return f"Dr. {owner}"
-        return str(owner)
-    return str(identity.get("name", "there"))
-
-
-def _customer_name(customer: Context) -> str:
-    name = str(customer.get("identity", {}).get("name") or "there")
-    return name.replace("(parent:", "parent:").strip()
-
-
-def _best_offer(merchant: Context, category: Context) -> str:
-    for offer in merchant.get("offers", []) or []:
-        if offer.get("status") == "active" and offer.get("title"):
-            return str(offer["title"])
-    catalog = [offer for offer in (category.get("offer_catalog", []) or []) if offer.get("title")]
-    preferred_types = {"service_at_price": 0, "free_service": 1, "membership": 2}
-    def rank(offer: Context) -> tuple[int, int]:
-        title = str(offer.get("title", "")).lower()
-        percent_penalty = 5 if "flat" in title and "%" in title else 0
-        return (preferred_types.get(str(offer.get("type")), 3) + percent_penalty, len(title))
-    for offer in sorted(catalog, key=rank):
-        title = str(offer["title"])
-        if not ("flat" in title.lower() and "%" in title):
-            return title
-    if catalog:
-        return str(sorted(catalog, key=rank)[0]["title"])
-    return "a simple service-price offer"
-
-
-def _digest_item(category: Context, item_id: str | None) -> Context:
-    digest = category.get("digest", []) or []
-    if item_id:
-        for item in digest:
-            if item.get("id") == item_id:
-                return item
-    return digest[0] if digest else {}
-
-
-def _first_fact(item: Context, agg: Context) -> str:
-    facts: list[str] = []
-    if item.get("trial_n"):
-        facts.append(f"{item['trial_n']}-patient trial")
-    if item.get("summary"):
-        match = re.search(r"(\d+(?:\.\d+)?%|\d+(?:\.\d+)?\s?mSv|\d+(?:\.\d+)?\s?credits?)", str(item["summary"]))
-        if match:
-            facts.append(match.group(1))
-    if agg.get("high_risk_adult_count"):
-        facts.append(f"{agg['high_risk_adult_count']} high-risk adults in your roster")
-    return "; ".join(facts) + "." if facts else ""
-
-
-def _cohort_phrase(agg: Context) -> str:
-    if agg.get("high_risk_adult_count"):
-        return f"your {agg['high_risk_adult_count']} high-risk adult patients"
-    if agg.get("chronic_rx_count"):
-        return f"your {agg['chronic_rx_count']} chronic-Rx customers"
-    if agg.get("total_active_members"):
-        return f"your {agg['total_active_members']} active members"
-    if agg.get("lapsed_90d_plus") or agg.get("lapsed_180d_plus"):
-        return "your lapsed-customer cohort"
-    return "your current customers"
-
-
-def _member_count(agg: Context) -> str:
-    if agg.get("total_active_members"):
-        return f"{agg['total_active_members']} active members"
-    if agg.get("total_unique_ytd"):
-        return f"{agg['total_unique_ytd']} customers"
-    return "existing customers"
-
-
-def _pct(value: Any) -> str:
-    try:
-        num = float(value)
-    except (TypeError, ValueError):
-        return "0%"
-    return f"{num * 100:.0f}%" if abs(num) <= 1 else f"{num:.0f}%"
-
-
-def _time_from_iso(value: str | None) -> str | None:
-    if not value:
-        return None
-    match = re.search(r"T(\d{2}):(\d{2})", value)
-    if not match:
-        return value
-    hour = int(match.group(1))
-    minute = match.group(2)
-    suffix = "am" if hour < 12 else "pm"
-    hour = hour if 1 <= hour <= 12 else abs(hour - 12) or 12
-    return f"{hour}:{minute}{suffix}"
-
-
-def _date_from_iso(value: str | None) -> str:
-    if not value:
-        return "the due date"
-    return value.split("T", 1)[0]
-
-
-def _months_between(start: str | None, end: str | None) -> str | None:
-    if not start or not end:
-        return None
-    try:
-        a = datetime.fromisoformat(start[:10])
-        b = datetime.fromisoformat(end[:10])
-    except ValueError:
-        return None
-    months = max(1, round((b - a).days / 30))
-    return f"{months} months"
-
-
-def _slot_text(slots: list[Context]) -> str:
-    labels = [str(s.get("label")) for s in slots if s.get("label")]
-    if not labels:
-        return "the next available slot"
-    if len(labels) == 1:
-        return labels[0]
-    return " or ".join(labels[:2])
-
-
-def _has_consent(customer: Context, trigger: Context) -> bool:
-    prefs = customer.get("preferences", {})
-    if prefs.get("reminder_opt_in") is False:
-        return False
-    scopes = set(customer.get("consent", {}).get("scope", []) or [])
-    kind = trigger.get("kind", "")
-    if kind in {"recall_due", "appointment_tomorrow"}:
-        return bool(scopes.intersection({"recall_reminders", "appointment_reminders"}))
-    if kind in {"chronic_refill_due"}:
-        return bool(scopes.intersection({"refill_reminders", "delivery_notifications"}))
-    if kind in {"customer_lapsed_hard", "customer_lapsed_soft"}:
-        return bool(scopes.intersection({"winback_offers", "renewal_reminders", "promotional_offers"}))
-    if kind in {"wedding_package_followup"}:
-        return "bridal_package_followup" in scopes
-    if kind in {"trial_followup"}:
-        return bool(scopes.intersection({"kids_program_updates", "program_updates", "appointment_reminders"}))
-    return bool(scopes)
-
-
-def _has_all(payload: Context, *fields: str) -> bool:
-    return all(payload.get(field) not in (None, "", []) for field in fields)
-
-
-def _has_any(payload: Context, *fields: str) -> bool:
-    return any(payload.get(field) not in (None, "", []) for field in fields)
-
-
-def _is_placeholder_payload(payload: Context) -> bool:
-    return payload.get("placeholder") is True
-
-
-def _dedupe_dr(value: str) -> str:
-    return re.sub(r"\bDr\.\s+Dr\.\s+", "Dr. ", value).strip()
-
-
-def _final_scrub(body: str) -> str:
-    body = _dedupe_dr(body)
-    body = body.replace("None", "the available context")
-    body = body.replace("baseline normal", "the recent baseline")
-    body = re.sub(r"\b(up|down|dropped|rose|increased)\s+0%\b", "changed in the latest context", body)
-    body = body.replace("the available context days", "a while")
-    body = body.replace("the available context km", "nearby")
-    body = body.replace("festival is the available context days away", "a festival window is coming up")
-    return _clean_multiline(body)
-
-
-def _short_id(value: str) -> str:
-    cleaned = re.sub(r"[^a-zA-Z0-9]+", "_", value).strip("_")
-    parts = cleaned.split("_")
-    return "_".join(parts[:4])[:36] or "x"
-
-
-def _action_scope(merchant: Context, trigger: Context) -> str:
-    name = merchant.get("identity", {}).get("name") or "this merchant"
-    kind = str(trigger.get("kind", "task")).replace("_", " ")
-    return f"{name}'s {kind}"
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+import hashlib
+import re
+
+from .decision_engine import compose_scored
+
+
+Context = dict[str, Any]
+
+
+CTA_NONE = "none"
+CTA_OPEN = "open_ended"
+CTA_YES_NO = "binary_yes_no"
+CTA_CONFIRM = "binary_confirm_cancel"
+CTA_SLOTS = "multi_choice_slot"
+
+
+def compose(category: Context, merchant: Context, trigger: Context, customer: Context | None = None) -> Context:
+    """Compose a deterministic, context-grounded Vera message."""
+    scored = compose_scored(category, merchant, trigger, customer)
+    if scored:
+        return validate_message(scored)
+    if customer:
+        return _compose_customer(category, merchant, trigger, customer)
+    return _compose_merchant(category, merchant, trigger)
+
+
+def reply_to_message(
+    message: str,
+    conversation: Context | None = None,
+    merchant: Context | None = None,
+    trigger: Context | None = None,
+) -> Context:
+    """Respond to a merchant/customer reply without using an LLM."""
+    text = (message or "").strip()
+    lower = text.lower()
+    conversation = conversation or {}
+    auto_count = int(conversation.get("auto_reply_count", 0))
+
+    if is_stop_or_hostile(lower):
+        return {
+            "action": "end",
+            "rationale": "The sender explicitly rejected further messages or used hostile stop language; ending without another nudge.",
+        }
+
+    if is_auto_reply(text):
+        if auto_count >= 3:
+            return {
+                "action": "end",
+                "rationale": "Repeated canned auto-reply detected three times; closing the conversation to avoid wasting turns.",
+            }
+        if auto_count == 2:
+            return {
+                "action": "wait",
+                "wait_seconds": 86400,
+                "rationale": "Same auto-reply repeated; owner is likely unavailable, so Vera waits 24 hours.",
+            }
+        return {
+            "action": "wait",
+            "wait_seconds": 14400,
+            "rationale": "Canned WhatsApp Business auto-reply detected; backing off 4 hours for a real owner reply.",
+        }
+
+    if is_commitment(lower):
+        scope = _action_scope(merchant or {}, trigger or {})
+        return {
+            "action": "send",
+            "body": f"Done. I am preparing {scope} now from the details already shared. I will keep it to one ready-to-send draft/action and avoid adding anything not in your current context.",
+            "cta": CTA_NONE,
+            "rationale": "The sender committed; switching directly to action mode without another qualification or confirmation loop.",
+        }
+
+    if is_offtopic(lower):
+        return {
+            "action": "send",
+            "body": "That is outside what I can help with directly. Coming back to this Vera task: should I prepare the draft/action from the details above?",
+            "cta": CTA_YES_NO,
+            "rationale": "Politely declines an off-topic request and returns to the active merchant-growth task.",
+        }
+
+    if is_delay(lower):
+        return {
+            "action": "wait",
+            "wait_seconds": 1800,
+            "rationale": "The sender asked for time or signaled they are busy; wait 30 minutes.",
+        }
+
+    if "?" in text or any(w in lower for w in ["what", "how", "price", "cost", "details", "send"]):
+        return {
+            "action": "send",
+            "body": "Yes. I can keep it simple: I will draft one ready-to-send version using only your current offer, locality, and the trigger we discussed. Reply YES and I will prepare it.",
+            "cta": CTA_YES_NO,
+            "rationale": "The sender is engaged and asking for details; answer briefly and request one low-friction confirmation.",
+        }
+
+    return {
+        "action": "send",
+        "body": "Got it. Should I prepare the next draft/action for this now?",
+        "cta": CTA_YES_NO,
+        "rationale": "Acknowledges an ambiguous but non-negative reply and asks for one clear next step.",
+    }
+
+
+def is_auto_reply(message: str) -> bool:
+    lower = (message or "").lower().strip()
+    patterns = [
+        "thank you for contacting",
+        "thanks for contacting",
+        "our team will respond",
+        "we will respond shortly",
+        "we are currently unavailable",
+        "business hours",
+        "automated assistant",
+        "auto-reply",
+        "away message",
+        "will get back to you",
+    ]
+    return any(p in lower for p in patterns)
+
+
+def is_stop_or_hostile(lower: str) -> bool:
+    if re.search(r"\b(stop|unsubscribe|remove me|opt out)\b", lower):
+        return True
+    hard_stops = [
+        "stop messaging",
+        "stop sending",
+        "unsubscribe",
+        "not interested",
+        "dont message",
+        "don't message",
+        "useless spam",
+        "spam",
+        "leave me",
+        "bothering me",
+    ]
+    return any(p in lower for p in hard_stops)
+
+
+def is_commitment(lower: str) -> bool:
+    commitments = [
+        r"\byes\b",
+        r"\bok\b",
+        r"\bokay\b",
+        r"\bgo ahead\b",
+        r"\blets do it\b",
+        r"\blet's do it\b",
+        r"\bconfirm\b",
+        r"\bproceed\b",
+        r"\bsend it\b",
+        r"\bdo it\b",
+        r"\bstart\b",
+        r"\bi want to join\b",
+        r"\bmujhe magicpin\b",
+        r"\bchalo\b",
+    ]
+    return any(re.search(pattern, lower) for pattern in commitments)
+
+
+def is_delay(lower: str) -> bool:
+    return any(p in lower for p in ["later", "busy", "after some time", "tomorrow", "call later", "not now"])
+
+
+def is_offtopic(lower: str) -> bool:
+    return any(p in lower for p in ["gst", "tax", "income tax", "hiring", "salary", "loan", "rent agreement", "website design", "cricket score", "weather", "movie ticket"])
+
+
+def validate_message(message: Context) -> Context:
+    """Normalize shape and remove risky leftovers."""
+    body = _clean_multiline(str(message.get("body", "")))
+    if not body:
+        body = "Quick Vera update: I found one relevant action from your current context. Want me to prepare it?"
+    cta = message.get("cta") or CTA_OPEN
+    if cta not in {CTA_NONE, CTA_OPEN, CTA_YES_NO, CTA_CONFIRM, CTA_SLOTS, "binary_yes_stop"}:
+        cta = CTA_OPEN
+    body = _final_scrub(body)
+    message["body"] = body[:1800]
+    message["cta"] = cta
+    message.setdefault("send_as", "vera")
+    message.setdefault("suppression_key", "")
+    message.setdefault("rationale", "Composed deterministically from category, merchant, trigger, and optional customer context.")
+    return message
+
+
+def make_conversation_id(merchant_id: str, trigger_id: str, customer_id: str | None = None) -> str:
+    base = f"{merchant_id}:{trigger_id}:{customer_id or ''}"
+    digest = hashlib.sha1(base.encode("utf-8")).hexdigest()[:8]
+    merchant_short = _short_id(merchant_id)
+    trigger_short = _short_id(trigger_id)
+    if customer_id:
+        return f"conv_{merchant_short}_{_short_id(customer_id)}_{digest}"
+    return f"conv_{merchant_short}_{trigger_short}_{digest}"
+
+
+def _compose_merchant(category: Context, merchant: Context, trigger: Context) -> Context:
+    kind = trigger.get("kind", "generic")
+    cat = merchant.get("category_slug") or category.get("slug", "merchant")
+    identity = merchant.get("identity", {})
+    name = _merchant_name(merchant)
+    first = _owner_or_name(merchant)
+    payload = trigger.get("payload", {})
+    offer = _best_offer(merchant, category)
+    perf = merchant.get("performance", {})
+    agg = merchant.get("customer_aggregate", {})
+    locality = identity.get("locality") or identity.get("city") or "your area"
+
+    if _is_placeholder_payload(payload):
+        return _generic_merchant(category, merchant, trigger)
+
+    if kind in {"research_digest", "regulation_change", "cde_opportunity"}:
+        item = _digest_item(category, payload.get("top_item_id") or payload.get("digest_item_id"))
+        title = item.get("title") or payload.get("metric_or_topic") or "this week's category update"
+        source = item.get("source") or "your category digest"
+        number = _first_fact(item, agg)
+        if kind == "regulation_change":
+            deadline = payload.get("deadline_iso") or trigger.get("expires_at", "")[:10]
+            body = f"{name}, compliance note: {title}. Deadline: {deadline}. {number} Source: {source}. Want me to draft the 5-point SOP/checklist for your clinic?"
+            cta = CTA_YES_NO
+        elif kind == "cde_opportunity":
+            credits = payload.get("credits") or item.get("credits")
+            fee = str(payload.get("fee") or item.get("actionable") or "").replace("_", " ")
+            body = f"{name}, IDA/CDE item for you: {title}. Credits: {credits}; fee: {fee}. Source: {source}. Want me to pull the invite and make a 1-line calendar reminder?"
+            cta = CTA_YES_NO
+        else:
+            cohort = _cohort_phrase(agg)
+            body = f"{name}, {source} has one item relevant to {cohort}: {title}. {number} Want me to pull the 2-min summary and draft a patient WhatsApp?"
+            cta = CTA_OPEN
+        return _msg(body, cta, "vera", trigger, f"{kind} uses digest source plus merchant cohort/performance context.")
+
+    if kind in {"perf_dip", "seasonal_perf_dip"}:
+        if kind == "perf_dip" and not _has_all(payload, "metric", "delta_pct", "window"):
+            return _generic_merchant(category, merchant, trigger)
+        metric = payload.get("metric", "performance")
+        delta = _pct(payload.get("delta_pct") or perf.get("delta_7d", {}).get(f"{metric}_pct"))
+        baseline = payload.get("vs_baseline")
+        seasonal = payload.get("is_expected_seasonal")
+        if seasonal:
+            body = f"{first}, your {metric} is down {delta} this {payload.get('window', 'week')}, but this matches the {payload.get('season_note', 'seasonal dip')} pattern. Do not over-spend ads now; focus on your {_member_count(agg)}. Want me to draft a retention nudge?"
+        else:
+            body = f"{first}, {metric} dropped {delta} in {payload.get('window', '7d')}; baseline was {baseline} and current calls are {perf.get('calls', 'lower than usual')}. {offer} is the quickest concrete hook. Want me to draft a recovery WhatsApp/GBP post?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Performance dip trigger; recommends one low-effort recovery action.")
+
+    if kind == "perf_spike":
+        if not _has_all(payload, "metric", "delta_pct", "window"):
+            return _generic_merchant(category, merchant, trigger)
+        metric = payload.get("metric", "calls")
+        delta = _pct(payload.get("delta_pct"))
+        driver = str(payload.get("likely_driver") or "recent profile activity").replace("_", " ")
+        body = f"{first}, {metric} is up {delta} in {payload.get('window', '7d')} vs baseline {payload.get('vs_baseline', 'normal')}. Likely driver: {driver}. Want me to turn this into a repeatable post for this week?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Performance spike trigger; converts a winning signal into a repeatable action.")
+
+    if kind in {"active_planning_intent"}:
+        topic = str(payload.get("intent_topic", "growth plan")).replace("_", " ")
+        last = payload.get("merchant_last_message", "")
+        if "thali" in topic:
+            body = f"{first}, based on your '{last}' message, here is a starter corporate thali structure: 10 thalis at the current {offer}, 25+ with free delivery, 50+ with a filter-coffee add-on. Want me to draft the 3-line outreach note?"
+        elif "kids_yoga" in topic:
+            body = f"{first}, for kids yoga summer camp, keep it simple: age 7-12, 4 weeks, 3 sessions/week, Saturday trial, and use your current {offer}. Want me to draft the GBP post plus parent WhatsApp?"
+        else:
+            body = f"{first}, you asked about {topic}. I can turn it into one concrete offer using {offer}, your {locality} locality, and your current profile. Want the ready draft?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Merchant already showed planning intent; moves directly to a concrete draft.")
+
+    if kind == "festival_upcoming":
+        if not _has_all(payload, "festival", "days_until"):
+            return _generic_merchant(category, merchant, trigger)
+        festival = payload.get("festival", "festival")
+        days = payload.get("days_until")
+        body = f"{first}, {festival} is {days} days away and {cat} is marked relevant for this beat. Your active hook is {offer}. Want me to prepare a festival post now and hold it for approval?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Festival trigger with category relevance and existing offer.")
+
+    if kind == "ipl_match_today":
+        if not _has_all(payload, "match", "venue", "match_time_iso"):
+            return _generic_merchant(category, merchant, trigger)
+        match = payload.get("match", "today's match")
+        venue = payload.get("venue", "the stadium")
+        time = _time_from_iso(payload.get("match_time_iso")) or "tonight"
+        weeknight = payload.get("is_weeknight")
+        advice = "push a delivery-only offer" if not weeknight else "run a quick pre-match dine-in/post"
+        body = f"{first}, {match} at {venue} starts {time}. Since this is {'not ' if not weeknight else ''}a weeknight match, {advice} using your active {offer}. Want me to draft the banner text and Insta story?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "IPL trigger interpreted with day context and current restaurant offer.")
+
+    if kind == "review_theme_emerged":
+        if not _has_all(payload, "theme", "occurrences_30d"):
+            return _generic_merchant(category, merchant, trigger)
+        theme = str(payload.get("theme", "review theme")).replace("_", " ")
+        count = payload.get("occurrences_30d")
+        quote = payload.get("common_quote")
+        body = f"{first}, {count} reviews in 30d now mention {theme}; one customer said '{quote}'. Want me to draft a short public reply pattern plus an ops checklist?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Review theme trigger turns repeated feedback into reply and ops action.")
+
+    if kind == "milestone_reached":
+        if not _has_all(payload, "metric", "value_now", "milestone_value"):
+            return _generic_merchant(category, merchant, trigger)
+        value = payload.get("value_now")
+        target = payload.get("milestone_value")
+        metric = str(payload.get("metric", "milestone")).replace("_", " ")
+        body = f"{first}, you are at {value} {metric}; {target} is close. Want me to draft a polite review-request WhatsApp for recent happy customers?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Milestone trigger nudges a review/request action tied to current count.")
+
+    if kind == "renewal_due":
+        if not _has_any(payload, "days_remaining", "renewal_amount") and not merchant.get("subscription", {}).get("days_remaining"):
+            return _generic_merchant(category, merchant, trigger)
+        days = payload.get("days_remaining") or merchant.get("subscription", {}).get("days_remaining")
+        amount = payload.get("renewal_amount")
+        body = f"{first}, your {payload.get('plan', 'plan')} renewal is due in {days} days for Rs {amount}. Calls are down {_pct(merchant.get('performance', {}).get('delta_7d', {}).get('calls_pct'))}; before renewal, want me to show the 3 fixes likely to recover calls?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Renewal trigger pairs deadline with current performance risk.")
+
+    if kind in {"winback_eligible", "dormant_with_vera"}:
+        if not _has_any(payload, "days_since_expiry", "days_since_last_merchant_message", "lapsed_customers_added_since_expiry"):
+            return _generic_merchant(category, merchant, trigger)
+        days = payload.get("days_since_expiry") or payload.get("days_since_last_merchant_message") or merchant.get("subscription", {}).get("days_since_expiry")
+        lapsed = payload.get("lapsed_customers_added_since_expiry") or agg.get("lapsed_90d_plus") or agg.get("lapsed_180d_plus")
+        body = f"{first}, it has been {days} days since the last active Vera/subscription moment. You now have {lapsed} lapsed customers/signals to recover. Want me to draft one winback message using {offer}?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Dormancy/winback trigger; restarts with one concrete recovery action.")
+
+    if kind == "supply_alert":
+        if not _has_all(payload, "molecule", "affected_batches"):
+            return _generic_merchant(category, merchant, trigger)
+        batches = ", ".join(payload.get("affected_batches", [])[:3])
+        molecule = payload.get("molecule", "medicine")
+        chronic = agg.get("chronic_rx_count") or agg.get("total_unique_ytd") or "repeat"
+        body = f"{first}, urgent stock alert: {molecule} batches {batches} from {payload.get('manufacturer', 'the manufacturer')}. You have {chronic} chronic-Rx customers in context. Want me to draft the replacement WhatsApp and counter checklist?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Supply alert trigger uses molecule, batch numbers, and pharmacy repeat-customer context.")
+
+    if kind == "category_seasonal":
+        if not payload.get("trends"):
+            return _generic_merchant(category, merchant, trigger)
+        trends = ", ".join(str(t).replace("_", " ") for t in payload.get("trends", [])[:4])
+        body = f"{first}, summer demand shift is visible: {trends}. Since shelf action is recommended, want me to draft a 10-item reorder/checklist plus WhatsApp note?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Seasonal pharmacy/category trigger maps trends to stock action.")
+
+    if kind == "gbp_unverified":
+        if not _has_all(payload, "verification_path", "estimated_uplift_pct"):
+            return _generic_merchant(category, merchant, trigger)
+        uplift = _pct(payload.get("estimated_uplift_pct"))
+        body = f"{first}, your GBP is still unverified; the available path is {payload.get('verification_path', 'verification')}. Verified profiles can unlock about {uplift} more visibility in this context. Want me to walk you through the 3-step verification?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "GBP verification trigger uses exact path and estimated uplift.")
+
+    if kind == "competitor_opened":
+        if not _has_all(payload, "competitor_name", "distance_km", "opened_date", "their_offer"):
+            return _generic_merchant(category, merchant, trigger)
+        body = f"{name}, new competitor signal: {payload.get('competitor_name')} opened {payload.get('distance_km')} km away on {payload.get('opened_date')} with {payload.get('their_offer')}. Your current hook is {offer}. Want me to draft a sharper local post?"
+        return _msg(body, CTA_YES_NO, "vera", trigger, "Competitor trigger uses named competitor, distance, date, and offer comparison.")
+
+    if kind == "curious_ask_due":
+        body = f"{first}, quick check: what service has been most asked-for this week at {identity.get('name', 'your business')}? I will turn your answer into a Google post and a 4-line WhatsApp reply. Takes 5 min."
+        return _msg(body, CTA_OPEN, "vera", trigger, "Curious-ask cadence asks the merchant for one useful signal.")
+
+    return _generic_merchant(category, merchant, trigger)
+
+
+def _compose_customer(category: Context, merchant: Context, trigger: Context, customer: Context) -> Context:
+    if not _has_consent(customer, trigger):
+        return _msg(
+            f"{_owner_or_name(merchant)}, {_customer_name(customer)} has a {str(trigger.get('kind', 'customer')).replace('_', ' ')} signal, but the saved opt-in scope does not clearly cover this outreach. Want me to prepare a consent-safe approval note first?",
+            CTA_YES_NO,
+            "vera",
+            trigger,
+            "Customer trigger routed to merchant because consent/preference does not clearly permit direct outreach.",
+        )
+
+    kind = trigger.get("kind", "customer_message")
+    payload = trigger.get("payload", {})
+    merchant_name = merchant.get("identity", {}).get("name", "your merchant")
+    owner = merchant.get("identity", {}).get("owner_first_name") or merchant_name
+    customer_name = _customer_name(customer)
+    offer = _best_offer(merchant, category)
+    lang = str(customer.get("identity", {}).get("language_pref", "english")).lower()
+
+    if _is_placeholder_payload(payload):
+        return _generic_customer(category, merchant, trigger, customer)
+
+    if kind == "recall_due":
+        if not _has_any(payload, "service_due", "due_date", "available_slots"):
+            return _generic_customer(category, merchant, trigger, customer)
+        slots = payload.get("available_slots", [])
+        slot_text = _slot_text(slots)
+        months = _months_between(payload.get("last_service_date"), payload.get("due_date")) or "6-month"
+        if "hi" in lang:
+            body = f"Hi {customer_name}, {merchant_name} here. It has been about {months} since your last visit; your {payload.get('service_due', 'recall').replace('_', ' ')} is due. Apke liye slots: {slot_text}. {offer}. Reply 1/2 for a slot, or suggest a time."
+        else:
+            body = f"Hi {customer_name}, {merchant_name} here. It has been about {months} since your last visit; your {payload.get('service_due', 'recall').replace('_', ' ')} is due. Slots available: {slot_text}. {offer}. Reply 1/2 for a slot, or suggest a time."
+        return _msg(body, CTA_SLOTS, "merchant_on_behalf", trigger, "Customer recall uses due date, slot options, offer, and language preference.")
+
+    if kind in {"customer_lapsed_hard", "customer_lapsed_soft"}:
+        days = payload.get("days_since_last_visit")
+        focus = str(payload.get("previous_focus") or customer.get("preferences", {}).get("training_focus") or "your goal").replace("_", " ")
+        elapsed = f"{days} days" if days is not None else "a while"
+        body = f"Hi {customer_name}, {owner} from {merchant_name} here. It has been {elapsed} since your last visit; no pressure. We can restart with {offer}, matched to {focus}. Reply YES and I will hold a no-commitment slot."
+        return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Customer lapse trigger uses days since visit, prior focus, and merchant offer.")
+
+    if kind == "appointment_tomorrow":
+        appointment = payload.get("appointment_time") or payload.get("slot_label") or _slot_text(payload.get("available_slots", []))
+        service = str(payload.get("service") or customer.get("relationship", {}).get("services_received", ["appointment"])[-1]).replace("_", " ")
+        body = f"Hi {customer_name}, {merchant_name} here. Reminder for your {service} appointment tomorrow: {appointment}. Reply YES to confirm or tell us if you need to reschedule."
+        return _msg(body, CTA_CONFIRM, "merchant_on_behalf", trigger, "Appointment reminder uses customer relationship and available appointment timing.")
+
+    if kind == "wedding_package_followup":
+        if not _has_any(payload, "wedding_date", "days_to_wedding", "next_step_window_open"):
+            return _generic_customer(category, merchant, trigger, customer)
+        days = payload.get("days_to_wedding")
+        body = f"Hi {customer_name}, {owner} from {merchant_name} here. {days} days to your wedding; this is the right window for {str(payload.get('next_step_window_open', 'skin prep')).replace('_', ' ')}. {offer}. Want me to block your preferred Saturday slot?"
+        return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Bridal follow-up uses wedding date window, relationship history, and offer.")
+
+    if kind == "trial_followup":
+        if not _has_any(payload, "trial_date", "next_session_options"):
+            return _generic_customer(category, merchant, trigger, customer)
+        slots = _slot_text(payload.get("next_session_options", []))
+        body = f"Hi {customer_name}, {owner} from {merchant_name} here. Thanks for trying the class on {payload.get('trial_date')}. Next available option: {slots}. Want me to reserve it?"
+        return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Trial follow-up uses trial date and next available session.")
+
+    if kind == "chronic_refill_due":
+        if not _has_any(payload, "molecule_list", "stock_runs_out_iso"):
+            return _generic_customer(category, merchant, trigger, customer)
+        meds = ", ".join(payload.get("molecule_list", []))
+        address_note = "We have your saved delivery address on file." if payload.get("delivery_address_saved") else "We can confirm the delivery address after your reply."
+        body = f"Namaste {customer_name}, {merchant_name} here. Your monthly medicines ({meds}) are due by {_date_from_iso(payload.get('stock_runs_out_iso'))}. {address_note} Reply CONFIRM and our pharmacist will verify stock and delivery details before preparing it."
+        return _msg(body, CTA_CONFIRM, "merchant_on_behalf", trigger, "Refill reminder uses molecule list, run-out date, delivery status, and pharmacy offer.")
+
+    return _msg(
+        f"Hi {customer_name}, {merchant_name} here. A quick update related to your last visit is ready. {offer}. Reply YES if you want the details.",
+        CTA_YES_NO,
+        "merchant_on_behalf",
+        trigger,
+        "Generic customer-scoped trigger with consent and merchant offer.",
+    )
+
+
+def _generic_merchant(category: Context, merchant: Context, trigger: Context) -> Context:
+    first = _owner_or_name(merchant)
+    kind = str(trigger.get("kind", "update")).replace("_", " ")
+    perf = merchant.get("performance", {})
+    offer = _best_offer(merchant, category)
+    views = perf.get("views")
+    calls = perf.get("calls")
+    identity = merchant.get("identity", {})
+    locality = identity.get("locality") or identity.get("city") or "your locality"
+    fact = f"Current 30d views: {views}; calls: {calls}." if views is not None and calls is not None else f"Locality: {locality}."
+    body = f"{first}, Vera found a {kind} signal for {_merchant_name(merchant)}. {fact} Best hook: {offer}. Want me to draft the next WhatsApp/GBP action?"
+    return _msg(body, CTA_YES_NO, "vera", trigger, "Fallback grounded in trigger kind, performance, and offer.")
+
+
+def _generic_customer(category: Context, merchant: Context, trigger: Context, customer: Context) -> Context:
+    customer_name = _customer_name(customer)
+    merchant_name = merchant.get("identity", {}).get("name", "your merchant")
+    kind = str(trigger.get("kind", "customer update")).replace("_", " ")
+    last_visit = customer.get("relationship", {}).get("last_visit")
+    state = str(customer.get("state", "active")).replace("_", " ")
+    offer = _best_offer(merchant, category)
+    visit_text = f" Last visit: {last_visit}." if last_visit else ""
+    body = f"Hi {customer_name}, {merchant_name} here. Quick {kind} update for you based on your {state} status.{visit_text} {offer}. Reply YES if you want me to hold the next step."
+    return _msg(body, CTA_YES_NO, "merchant_on_behalf", trigger, "Customer fallback grounded in relationship state, last visit, and available offer.")
+
+
+def _msg(body: str, cta: str, send_as: str, trigger: Context, rationale: str) -> Context:
+    return validate_message({
+        "body": body,
+        "cta": cta,
+        "send_as": send_as,
+        "suppression_key": trigger.get("suppression_key", trigger.get("id", "")),
+        "rationale": rationale,
+    })
+
+
+def _clean(text: str) -> str:
+    return re.sub(r"\s+", " ", text).strip()
+
+
+def _clean_multiline(text: str) -> str:
+    text = re.sub(r"\.{2,}", ".", text)
+    text = re.sub(r";\s*\.", ".", text)
+    text = re.sub(r"[ \t]+", " ", text)
+    text = re.sub(r" *\n *", "\n", text)
+    text = re.sub(r"\n{3,}", "\n\n", text)
+    return text.strip()
+
+
+def _merchant_name(merchant: Context) -> str:
+    identity = merchant.get("identity", {})
+    name = identity.get("name") or "Merchant"
+    owner = str(identity.get("owner_first_name") or "").strip()
+    if merchant.get("category_slug") == "dentists" and owner and not str(name).lower().startswith("dr."):
+        return owner if owner.lower().startswith("dr") else f"Dr. {owner}"
+    return _dedupe_dr(str(name))
+
+
+def _owner_or_name(merchant: Context) -> str:
+    identity = merchant.get("identity", {})
+    owner = identity.get("owner_first_name")
+    if owner:
+        if merchant.get("category_slug") == "dentists" and not str(owner).lower().startswith("dr"):
+            return f"Dr. {owner}"
+        return str(owner)
+    return str(identity.get("name", "there"))
+
+
+def _customer_name(customer: Context) -> str:
+    name = str(customer.get("identity", {}).get("name") or "there")
+    return name.replace("(parent:", "parent:").strip()
+
+
+def _best_offer(merchant: Context, category: Context) -> str:
+    for offer in merchant.get("offers", []) or []:
+        if offer.get("status") == "active" and offer.get("title"):
+            return str(offer["title"])
+    catalog = [offer for offer in (category.get("offer_catalog", []) or []) if offer.get("title")]
+    preferred_types = {"service_at_price": 0, "free_service": 1, "membership": 2}
+    def rank(offer: Context) -> tuple[int, int]:
+        title = str(offer.get("title", "")).lower()
+        percent_penalty = 5 if "flat" in title and "%" in title else 0
+        return (preferred_types.get(str(offer.get("type")), 3) + percent_penalty, len(title))
+    for offer in sorted(catalog, key=rank):
+        title = str(offer["title"])
+        if not ("flat" in title.lower() and "%" in title):
+            return title
+    if catalog:
+        return str(sorted(catalog, key=rank)[0]["title"])
+    return "a simple service-price offer"
+
+
+def _digest_item(category: Context, item_id: str | None) -> Context:
+    digest = category.get("digest", []) or []
+    if item_id:
+        for item in digest:
+            if item.get("id") == item_id:
+                return item
+    return digest[0] if digest else {}
+
+
+def _first_fact(item: Context, agg: Context) -> str:
+    facts: list[str] = []
+    if item.get("trial_n"):
+        facts.append(f"{item['trial_n']}-patient trial")
+    if item.get("summary"):
+        match = re.search(r"(\d+(?:\.\d+)?%|\d+(?:\.\d+)?\s?mSv|\d+(?:\.\d+)?\s?credits?)", str(item["summary"]))
+        if match:
+            facts.append(match.group(1))
+    if agg.get("high_risk_adult_count"):
+        facts.append(f"{agg['high_risk_adult_count']} high-risk adults in your roster")
+    return "; ".join(facts) + "." if facts else ""
+
+
+def _cohort_phrase(agg: Context) -> str:
+    if agg.get("high_risk_adult_count"):
+        return f"your {agg['high_risk_adult_count']} high-risk adult patients"
+    if agg.get("chronic_rx_count"):
+        return f"your {agg['chronic_rx_count']} chronic-Rx customers"
+    if agg.get("total_active_members"):
+        return f"your {agg['total_active_members']} active members"
+    if agg.get("lapsed_90d_plus") or agg.get("lapsed_180d_plus"):
+        return "your lapsed-customer cohort"
+    return "your current customers"
+
+
+def _member_count(agg: Context) -> str:
+    if agg.get("total_active_members"):
+        return f"{agg['total_active_members']} active members"
+    if agg.get("total_unique_ytd"):
+        return f"{agg['total_unique_ytd']} customers"
+    return "existing customers"
+
+
+def _pct(value: Any) -> str:
+    try:
+        num = float(value)
+    except (TypeError, ValueError):
+        return "0%"
+    return f"{num * 100:.0f}%" if abs(num) <= 1 else f"{num:.0f}%"
+
+
+def _time_from_iso(value: str | None) -> str | None:
+    if not value:
+        return None
+    match = re.search(r"T(\d{2}):(\d{2})", value)
+    if not match:
+        return value
+    hour = int(match.group(1))
+    minute = match.group(2)
+    suffix = "am" if hour < 12 else "pm"
+    hour = hour if 1 <= hour <= 12 else abs(hour - 12) or 12
+    return f"{hour}:{minute}{suffix}"
+
+
+def _date_from_iso(value: str | None) -> str:
+    if not value:
+        return "the due date"
+    return value.split("T", 1)[0]
+
+
+def _months_between(start: str | None, end: str | None) -> str | None:
+    if not start or not end:
+        return None
+    try:
+        a = datetime.fromisoformat(start[:10])
+        b = datetime.fromisoformat(end[:10])
+    except ValueError:
+        return None
+    months = max(1, round((b - a).days / 30))
+    return f"{months} months"
+
+
+def _slot_text(slots: list[Context]) -> str:
+    labels = [str(s.get("label")) for s in slots if s.get("label")]
+    if not labels:
+        return "the next available slot"
+    if len(labels) == 1:
+        return labels[0]
+    return " or ".join(labels[:2])
+
+
+def _has_consent(customer: Context, trigger: Context) -> bool:
+    prefs = customer.get("preferences", {})
+    if prefs.get("reminder_opt_in") is False:
+        return False
+    scopes = set(customer.get("consent", {}).get("scope", []) or [])
+    kind = trigger.get("kind", "")
+    if kind in {"recall_due", "appointment_tomorrow"}:
+        return bool(scopes.intersection({"recall_reminders", "appointment_reminders"}))
+    if kind in {"chronic_refill_due"}:
+        return bool(scopes.intersection({"refill_reminders", "delivery_notifications"}))
+    if kind in {"customer_lapsed_hard", "customer_lapsed_soft"}:
+        return bool(scopes.intersection({"winback_offers", "renewal_reminders", "promotional_offers"}))
+    if kind in {"wedding_package_followup"}:
+        return "bridal_package_followup" in scopes
+    if kind in {"trial_followup"}:
+        return bool(scopes.intersection({"kids_program_updates", "program_updates", "appointment_reminders"}))
+    return bool(scopes)
+
+
+def _has_all(payload: Context, *fields: str) -> bool:
+    return all(payload.get(field) not in (None, "", []) for field in fields)
+
+
+def _has_any(payload: Context, *fields: str) -> bool:
+    return any(payload.get(field) not in (None, "", []) for field in fields)
+
+
+def _is_placeholder_payload(payload: Context) -> bool:
+    return payload.get("placeholder") is True
+
+
+def _dedupe_dr(value: str) -> str:
+    return re.sub(r"\bDr\.\s+Dr\.\s+", "Dr. ", value).strip()
+
+
+def _final_scrub(body: str) -> str:
+    body = _dedupe_dr(body)
+    body = body.replace("None", "the available context")
+    body = body.replace("baseline normal", "the recent baseline")
+    body = re.sub(r"\b(up|down|dropped|rose|increased)\s+0%\b", "changed in the latest context", body)
+    body = body.replace("the available context days", "a while")
+    body = body.replace("the available context km", "nearby")
+    body = body.replace("festival is the available context days away", "a festival window is coming up")
+    return _clean_multiline(body)
+
+
+def _short_id(value: str) -> str:
+    cleaned = re.sub(r"[^a-zA-Z0-9]+", "_", value).strip("_")
+    parts = cleaned.split("_")
+    return "_".join(parts[:4])[:36] or "x"
+
+
+def _action_scope(merchant: Context, trigger: Context) -> str:
+    name = merchant.get("identity", {}).get("name") or "this merchant"
+    kind = str(trigger.get("kind", "task")).replace("_", " ")
+    return f"{name}'s {kind}"

app/main.py

@@ -90,6 +90,7 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
     )
 
 
+@app.get("/healthz", include_in_schema=False)
 @app.get("/v1/healthz")
 async def healthz() -> dict[str, Any]:
     counts = {scope: 0 for scope in VALID_SCOPES}
@@ -103,6 +104,7 @@ async def healthz() -> dict[str, Any]:
     }
 
 
+@app.get("/metadata", include_in_schema=False)
 @app.get("/v1/metadata")
 async def metadata() -> dict[str, Any]:
     return {
@@ -119,6 +121,7 @@ async def metadata() -> dict[str, Any]:
     }
 
 
+@app.post("/context", include_in_schema=False)
 @app.post("/v1/context")
 async def push_context(body: ContextBody):
     if body.scope not in VALID_SCOPES:
@@ -146,6 +149,7 @@ async def push_context(body: ContextBody):
     return {"accepted": True, "ack_id": f"ack_{body.context_id}_v{body.version}", "stored_at": contexts[key]["stored_at"]}
 
 
+@app.post("/tick", include_in_schema=False)
 @app.post("/v1/tick")
 async def tick(body: TickBody) -> dict[str, Any]:
     actions: list[dict[str, Any]] = []
@@ -228,6 +232,7 @@ async def tick(body: TickBody) -> dict[str, Any]:
     return {"actions": actions}
 
 
+@app.post("/reply", include_in_schema=False)
 @app.post("/v1/reply")
 async def reply(body: ReplyBody) -> dict[str, Any]:
     conv = conversations.setdefault(
@@ -296,6 +301,7 @@ async def reply(body: ReplyBody) -> dict[str, Any]:
     return result
 
 
+@app.post("/teardown", include_in_schema=False)
 @app.post("/v1/teardown")
 async def teardown() -> dict[str, Any]:
     contexts.clear()

tests/test_bot.py

@@ -1,585 +1,585 @@
-from __future__ import annotations
-
-import json
-from pathlib import Path
-import sys
-
-from fastapi.testclient import TestClient
-
-ROOT = Path(__file__).resolve().parents[1]
-sys.path.insert(0, str(ROOT))
-
-from app.decision_engine import build_candidates, build_merchant_car, case_similarity, constitutional_violations, extract_evidence, retrieve_similar_case, score_map, select_cialdini_principle  # noqa: E402
-from app.main import app, category_arm_pool, contexts, conversations, merchant_action_memory, merchant_auto_replies, merchant_opt_out, suppressed  # noqa: E402
-
-
-def load_seed(name: str, key: str) -> list[dict]:
-    return json.loads((ROOT / "dataset" / name).read_text(encoding="utf-8"))[key]
-
-
-def load_category(slug: str) -> dict:
-    return json.loads((ROOT / "dataset" / "categories" / f"{slug}.json").read_text(encoding="utf-8"))
-
-
-def reset_state() -> None:
-    contexts.clear()
-    conversations.clear()
-    suppressed.clear()
-    merchant_opt_out.clear()
-    merchant_auto_replies.clear()
-    merchant_action_memory.clear()
-    category_arm_pool.clear()
-
-
-def push(client: TestClient, scope: str, context_id: str, payload: dict, version: int = 1):
-    return client.post(
-        "/v1/context",
-        json={"scope": scope, "context_id": context_id, "version": version, "payload": payload, "delivered_at": "2026-04-26T10:00:00Z"},
-    )
-
-
-def test_health_and_context_idempotency():
-    reset_state()
-    client = TestClient(app)
-    assert client.get("/v1/healthz").json()["status"] == "ok"
-    cat = load_category("dentists")
-    resp = push(client, "category", "dentists", cat)
-    assert resp.status_code == 200
-    assert resp.json()["accepted"] is True
-    same = push(client, "category", "dentists", cat)
-    assert same.status_code == 200
-    assert same.json()["idempotent"] is True
-    stale = push(client, "category", "dentists", cat, version=0)
-    assert stale.status_code == 409
-
-
-def test_tick_creates_grounded_research_action():
-    reset_state()
-    client = TestClient(app)
-    merchants = load_seed("merchants_seed.json", "merchants")
-    triggers = load_seed("triggers_seed.json", "triggers")
-    merchant = merchants[0]
-    trigger = triggers[0]
-    push(client, "category", "dentists", load_category("dentists"))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "trigger", trigger["id"], trigger)
-
-    resp = client.post("/v1/tick", json={"now": "2026-04-26T10:30:00Z", "available_triggers": [trigger["id"]]})
-    assert resp.status_code == 200
-    actions = resp.json()["actions"]
-    assert len(actions) == 1
-    body = actions[0]["body"]
-    assert "JIDA" in body
-    assert "124" in body
-    assert actions[0]["send_as"] == "vera"
-    assert actions[0]["suppression_key"] == trigger["suppression_key"]
-
-
-def test_customer_consent_and_send_as():
-    reset_state()
-    client = TestClient(app)
-    merchants = load_seed("merchants_seed.json", "merchants")
-    customers = load_seed("customers_seed.json", "customers")
-    triggers = load_seed("triggers_seed.json", "triggers")
-    merchant = merchants[0]
-    customer = customers[0]
-    trigger = triggers[2]
-    push(client, "category", "dentists", load_category("dentists"))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "customer", customer["customer_id"], customer)
-    push(client, "trigger", trigger["id"], trigger)
-
-    resp = client.post("/v1/tick", json={"now": "2026-04-26T11:00:00Z", "available_triggers": [trigger["id"]]})
-    action = resp.json()["actions"][0]
-    assert action["send_as"] == "merchant_on_behalf"
-    assert action["customer_id"] == customer["customer_id"]
-    assert "Priya" in action["body"]
-    assert "Wed 5 Nov" in action["body"]
-
-
-def test_reply_replay_behaviors():
-    reset_state()
-    client = TestClient(app)
-    auto = "Thank you for contacting us! Our team will respond shortly."
-    first = client.post("/v1/reply", json={"conversation_id": "conv_auto", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 2}).json()
-    assert first["action"] == "wait"
-    second = client.post("/v1/reply", json={"conversation_id": "conv_auto", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 3}).json()
-    assert second["action"] == "wait"
-    third = client.post("/v1/reply", json={"conversation_id": "conv_auto", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 4}).json()
-    assert third["action"] == "end"
-
-    intent = client.post("/v1/reply", json={"conversation_id": "conv_intent", "merchant_id": "m1", "from_role": "merchant", "message": "Ok lets do it. Whats next?", "turn_number": 2}).json()
-    assert intent["action"] == "send"
-    assert "preparing" in intent["body"]
-    assert intent["cta"] == "none"
-
-    hostile = client.post("/v1/reply", json={"conversation_id": "conv_hostile", "merchant_id": "m1", "from_role": "merchant", "message": "Stop messaging me. This is useless spam.", "turn_number": 2}).json()
-    assert hostile["action"] == "end"
-
-    plain_stop = client.post("/v1/reply", json={"conversation_id": "conv_plain_stop", "merchant_id": "m1", "from_role": "merchant", "message": "STOP", "turn_number": 2}).json()
-    assert plain_stop["action"] == "end"
-
-    offtopic = client.post("/v1/reply", json={"conversation_id": "conv_offtopic", "merchant_id": "m1", "from_role": "merchant", "message": "What is the cricket score?", "turn_number": 2}).json()
-    assert offtopic["action"] == "send"
-    assert "outside" in offtopic["body"].lower()
-
-
-def test_ended_conversation_never_sends_again_and_omitted_merchant_optout():
-    reset_state()
-    client = TestClient(app)
-    conversations["conv_done"] = {"merchant_id": "m_done", "ended": True, "turns": []}
-    resp = client.post("/v1/reply", json={"conversation_id": "conv_done", "from_role": "merchant", "message": "hello", "turn_number": 5}).json()
-    assert resp["action"] == "end"
-
-    conversations["conv_stop"] = {"merchant_id": "m_stop", "ended": False, "turns": []}
-    stop = client.post("/v1/reply", json={"conversation_id": "conv_stop", "from_role": "merchant", "message": "Stop messaging me", "turn_number": 2}).json()
-    assert stop["action"] == "end"
-    assert "m_stop" in merchant_opt_out
-
-
-def test_validation_errors_are_challenge_style_400():
-    reset_state()
-    client = TestClient(app)
-    resp = client.post("/v1/context", json={"scope": "category"})
-    assert resp.status_code == 400
-    body = resp.json()
-    assert body["accepted"] is False
-    assert body["reason"] == "malformed"
-
-
-def test_merchant_level_auto_reply_tracking_across_conversations():
-    reset_state()
-    client = TestClient(app)
-    auto = "Thank you for contacting us! Our team will respond shortly."
-    assert client.post("/v1/reply", json={"conversation_id": "conv_auto_1", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 2}).json()["action"] == "wait"
-    assert client.post("/v1/reply", json={"conversation_id": "conv_auto_2", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 3}).json()["action"] == "wait"
-    assert client.post("/v1/reply", json={"conversation_id": "conv_auto_3", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 4}).json()["action"] == "end"
-
-
-def test_placeholder_triggers_do_not_leak_missing_fields():
-    reset_state()
-    client = TestClient(app)
-    merchant = load_seed("merchants_seed.json", "merchants")[5]
-    trigger = {
-        "id": "trg_placeholder_competitor",
-        "scope": "merchant",
-        "kind": "competitor_opened",
-        "source": "external",
-        "merchant_id": merchant["merchant_id"],
-        "customer_id": None,
-        "payload": {"placeholder": True, "metric_or_topic": "competitor_opened"},
-        "urgency": 2,
-        "suppression_key": "competitor:placeholder",
-        "expires_at": "2026-06-30T00:00:00Z",
-    }
-    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "trigger", trigger["id"], trigger)
-    action = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
-    assert "None" not in action["body"]
-    assert "the available context" not in action["body"]
-    assert action["cta"] == "binary_yes_no"
-
-
-def test_customer_without_matching_consent_routes_to_merchant_not_dead_line():
-    reset_state()
-    client = TestClient(app)
-    merchant = load_seed("merchants_seed.json", "merchants")[7]
-    customer = load_seed("customers_seed.json", "customers")[11]
-    trigger = {
-        "id": "trg_recall_no_scope",
-        "scope": "customer",
-        "kind": "recall_due",
-        "source": "internal",
-        "merchant_id": merchant["merchant_id"],
-        "customer_id": customer["customer_id"],
-        "payload": {"placeholder": True, "metric_or_topic": "recall_due"},
-        "urgency": 2,
-        "suppression_key": "recall:no_scope",
-        "expires_at": "2026-06-30T00:00:00Z",
-    }
-    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "customer", customer["customer_id"], customer)
-    push(client, "trigger", trigger["id"], trigger)
-    action = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
-    assert action["send_as"] == "vera"
-    assert action["cta"] == "binary_yes_no"
-    assert "I will not send" not in action["body"]
-
-
-def test_decision_engine_scores_case_anchor_shapes():
-    merchants = load_seed("merchants_seed.json", "merchants")
-    customers = load_seed("customers_seed.json", "customers")
-    triggers = load_seed("triggers_seed.json", "triggers")
-    cases = [
-        (merchants[0], triggers[0], None),
-        (merchants[0], triggers[2], customers[0]),
-        (merchants[4], triggers[9], None),
-        (merchants[6], triggers[13], None),
-        (merchants[8], triggers[17], None),
-        (merchants[8], triggers[18], customers[12]),
-    ]
-    for merchant, trigger, customer in cases:
-        category = load_category(merchant["category_slug"])
-        evidence = extract_evidence(category, merchant, trigger, customer)
-        candidates = build_candidates(category, merchant, trigger, customer, evidence)
-        assert candidates
-        best = max(candidates, key=lambda c: c.total_score)
-        assert best.total_score >= 36, (trigger["id"], best.total_score, best.body, best.rubric_scores)
-        assert any(ch.isdigit() for ch in best.body)
-        assert trigger["kind"].replace("_", " ") in best.body.lower() or any(e.source.startswith("trigger") for e in best.evidence)
-
-
-def test_full_expanded_dataset_proxy_has_no_weak_outputs():
-    import subprocess
-    import sys as _sys
-    subprocess.run([_sys.executable, "dataset/generate_dataset.py", "--seed-dir", "dataset", "--out", "expanded"], cwd=ROOT, check=True, capture_output=True)
-    result = subprocess.run([_sys.executable, "scripts/score_proxy.py", "34"], cwd=ROOT, text=True, capture_output=True)
-    assert result.returncode == 0, result.stdout + result.stderr
-
-
-def test_car_map_jitai_and_best_of_n_debug_fields():
-    merchant = load_seed("merchants_seed.json", "merchants")[0]
-    trigger = load_seed("triggers_seed.json", "triggers")[0]
-    category = load_category(merchant["category_slug"])
-    car = build_merchant_car(category, merchant, trigger)
-    assert car.merchant_name != "unknown"
-    assert car.category == "dentists"
-    assert all(value is not None for value in car.summary().values())
-
-    evidence = extract_evidence(category, merchant, trigger, None, car)
-    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
-    assert len(candidates) >= 3
-    best = max(candidates, key=lambda c: c.total_score)
-    assert best.decision_plan if hasattr(best, "decision_plan") else True
-    assert best.car_summary["category"] == "dentists"
-    assert {"severity", "receptivity", "intervention_fit"} <= set(best.jitai_scores)
-    assert {"motivation", "ability", "prompt"} <= set(best.map_scores)
-    assert best.frame in {"loss_frame", "gain_frame", "certainty_frame", "social_proof", "professional_value", "effort_externalization"}
-
-
-def test_bmap_penalizes_high_friction_cta():
-    merchant = load_seed("merchants_seed.json", "merchants")[0]
-    trigger = load_seed("triggers_seed.json", "triggers")[0]
-    category = load_category(merchant["category_slug"])
-    car = build_merchant_car(category, merchant, trigger)
-    easy = score_map(car, trigger, "JIDA says 124 patients are relevant. Reply YES and I will draft it.", "binary_yes_no", "professional_value", [])
-    hard = score_map(car, trigger, "JIDA says 124 patients are relevant. Please call, log in, choose a campaign, upload a file, and configure delivery.", "open_ended", "professional_value", [])
-    assert easy["ability"] > hard["ability"]
-
-
-def test_frames_follow_trigger_shape_and_action_memory_changes_plan():
-    merchant = load_seed("merchants_seed.json", "merchants")[1]
-    category = load_category(merchant["category_slug"])
-    dip = load_seed("triggers_seed.json", "triggers")[3]
-    spike = {**dip, "id": "trg_spike_test", "kind": "perf_spike", "payload": {"metric": "calls", "delta_pct": 0.18}, "urgency": 2}
-    dip_candidates = build_candidates(category, merchant, dip, None, extract_evidence(category, merchant, dip))
-    spike_candidates = build_candidates(category, merchant, spike, None, extract_evidence(category, merchant, spike))
-    assert any(c.frame == "loss_frame" for c in dip_candidates)
-    assert any(c.frame == "gain_frame" for c in spike_candidates)
-
-    remembered = {**merchant, "__vera_memory": {"last_action_type": "recovery_nudge", "last_response_intent": "auto_reply", "repeated_action_count": 3, "no_reply_count": 2}}
-    car = build_merchant_car(category, remembered, dip)
-    evidence = extract_evidence(category, remembered, dip, None, car)
-    candidates = build_candidates(category, remembered, dip, None, evidence, car)
-    assert all(c.jitai_scores["receptivity"] <= 4 for c in candidates)
-
-
-def test_openrouter_calibration_skips_without_key(monkeypatch):
-    import subprocess
-    import sys as _sys
-    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
-    result = subprocess.run([_sys.executable, "scripts/geval_calibrate.py"], cwd=ROOT, text=True, capture_output=True)
-    assert result.returncode == 0
-    assert "skipped" in result.stdout.lower()
-
-
-def test_cialdini_constitution_and_tot_debug_fields():
-    merchant = load_seed("merchants_seed.json", "merchants")[0]
-    trigger = load_seed("triggers_seed.json", "triggers")[0]
-    category = load_category(merchant["category_slug"])
-    car = build_merchant_car(category, merchant, trigger)
-    evidence = extract_evidence(category, merchant, trigger, None, car)
-    principle = select_cialdini_principle(car, trigger, evidence, "professional_value")
-    assert principle in {"authority", "social_proof", "liking", "reciprocity", "scarcity", "commitment"}
-    bad = "Dear valued partner, want to increase sales? Contact us?"
-    violations = constitutional_violations(bad, car, trigger, "open_ended")
-    assert "generic_or_corporate_copy" in violations
-
-    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
-    best = max(candidates, key=lambda c: c.total_score)
-    assert best.thought_frames
-    assert best.persuasion_principle
-    assert best.reference_key.startswith("dentists:")
-    assert not best.constitutional_violations
-
-
-def test_category_empirical_prior_flows_into_car():
-    reset_state()
-    client = TestClient(app)
-    merchant = load_seed("merchants_seed.json", "merchants")[0]
-    trigger = load_seed("triggers_seed.json", "triggers")[0]
-    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "trigger", trigger["id"], trigger)
-    first = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
-    client.post("/v1/reply", json={"conversation_id": first["conversation_id"], "merchant_id": merchant["merchant_id"], "from_role": "merchant", "message": "yes go ahead", "turn_number": 2})
-    assert category_arm_pool[merchant["category_slug"]]
-    next_trigger = {**trigger, "id": "trg_research_next", "suppression_key": "research:next"}
-    push(client, "trigger", next_trigger["id"], next_trigger, version=1)
-    second = client.post("/v1/tick", json={"now": "2026-04-26T10:05:00Z", "available_triggers": [next_trigger["id"]]}).json()["actions"][0]
-    priors = second["decision_plan"]["car_summary"]["category_arm_priors"]
-    assert priors
-
-
-def test_sparse_context_fallback_stays_specific_and_safe():
-    category = load_category("restaurants")
-    merchant = {
-        "merchant_id": "m_sparse_restaurant",
-        "category_slug": "restaurants",
-        "identity": {"name": "Asha Cafe", "owner_first_name": "Asha", "locality": "Indiranagar"},
-        "performance": {},
-        "offers": [],
-        "customer_aggregate": {},
-        "signals": [],
-        "conversation_history": [],
-    }
-    trigger = {
-        "id": "trg_sparse_reactivation",
-        "scope": "merchant",
-        "kind": "merchant_inactive",
-        "source": "internal",
-        "merchant_id": merchant["merchant_id"],
-        "payload": {"days_inactive": 14},
-        "urgency": 1,
-        "expires_at": "2026-06-30T00:00:00Z",
-    }
-    car = build_merchant_car(category, merchant, trigger)
-    evidence = extract_evidence(category, merchant, trigger, None, car)
-    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
-    assert len(candidates) == 1
-    body = candidates[0].body
-    assert "Asha" in body
-    assert "Indiranagar" in body
-    assert "increase sales" not in body.lower()
-    assert "sparse_context_floor" in candidates[0].risk_flags
-
-
-def test_broad_suppression_keys_are_made_unique():
-    reset_state()
-    client = TestClient(app)
-    merchant = load_seed("merchants_seed.json", "merchants")[4]
-    category = load_category(merchant["category_slug"])
-    push(client, "category", merchant["category_slug"], category)
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    t1 = {"id": "trg_broad_1", "scope": "merchant", "kind": "curious_ask_due", "source": "internal", "merchant_id": merchant["merchant_id"], "payload": {"topic": "lunch"}, "urgency": 2, "suppression_key": "curious_ask_due"}
-    t2 = {"id": "trg_broad_2", "scope": "merchant", "kind": "curious_ask_due", "source": "internal", "merchant_id": merchant["merchant_id"], "payload": {"topic": "dinner"}, "urgency": 2, "suppression_key": "curious_ask_due"}
-    push(client, "trigger", t1["id"], t1)
-    push(client, "trigger", t2["id"], t2)
-    actions = client.post("/v1/tick", json={"now": "2026-05-02T10:00:00Z", "available_triggers": [t1["id"], t2["id"]]}).json()["actions"]
-    assert len(actions) == 2
-    assert actions[0]["suppression_key"] != actions[1]["suppression_key"]
-
-
-def test_context_updates_do_not_wipe_reply_memory():
-    reset_state()
-    client = TestClient(app)
-    merchant = load_seed("merchants_seed.json", "merchants")[0]
-    trigger = load_seed("triggers_seed.json", "triggers")[0]
-    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "trigger", trigger["id"], trigger)
-    first = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
-    client.post("/v1/reply", json={"conversation_id": first["conversation_id"], "merchant_id": merchant["merchant_id"], "from_role": "merchant", "message": "yes go ahead"})
-    push(client, "merchant", merchant["merchant_id"], {**merchant, "signals": ["fresh update"]}, version=2)
-    assert merchant_action_memory[merchant["merchant_id"]]["last_response_intent"] == "commitment"
-
-
-def test_pharmacy_without_consent_routes_to_merchant_and_avoids_medical_dispatch():
-    merchant = load_seed("merchants_seed.json", "merchants")[8]
-    customer = {**load_seed("customers_seed.json", "customers")[12], "consent": {"scope": []}}
-    trigger = load_seed("triggers_seed.json", "triggers")[18]
-    category = load_category(merchant["category_slug"])
-    car = build_merchant_car(category, merchant, trigger, customer)
-    evidence = extract_evidence(category, merchant, trigger, customer, car)
-    candidates = build_candidates(category, merchant, trigger, customer, evidence, car)
-    best = max(candidates, key=lambda c: c.total_score)
-    assert best.send_as == "vera"
-    assert "consent-safe" in best.body
-    assert "dispatch" not in best.body.lower()
-    assert "pharmacy_consent_or_medical_advice_risk" not in best.constitutional_violations
-
-
-def test_unseen_ops_trigger_uses_payload_without_known_template():
-    reset_state()
-    client = TestClient(app)
-    merchant = load_seed("merchants_seed.json", "merchants")[4]
-    trigger = {
-        "id": "trg_unseen_delivery_delay",
-        "scope": "merchant",
-        "kind": "delivery_delay_spike",
-        "source": "review_digest",
-        "merchant_id": merchant["merchant_id"],
-        "payload": {
-            "theme": "delivery_late",
-            "occurrences_30d": 6,
-            "avg_delay_minutes": 42,
-            "metric": "late delivery complaints",
-        },
-        "urgency": 3,
-        "suppression_key": "delivery-delay:unseen",
-        "expires_at": "2026-06-30T00:00:00Z",
-    }
-    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "trigger", trigger["id"], trigger)
-    action = client.post("/v1/tick", json={"now": "2026-05-02T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
-    body = action["body"].lower()
-    assert "42" in body
-    assert "delivery late" in body or "late delivery" in body
-    assert "prep/rider handoff" in body
-    assert "increase sales" not in body
-
-
-def test_missing_digest_id_retrieves_nearest_category_fact_for_research():
-    merchant = load_seed("merchants_seed.json", "merchants")[0]
-    category = load_category("dentists")
-    trigger = {
-        "id": "trg_unseen_aligner_research",
-        "scope": "merchant",
-        "kind": "research_digest",
-        "source": "external_digest",
-        "merchant_id": merchant["merchant_id"],
-        "payload": {
-            "top_item_id": "d_not_in_seed",
-            "topic": "clear aligner consultations",
-            "metric_or_topic": "clear aligners near me",
-        },
-        "urgency": 2,
-        "suppression_key": "research:unseen-aligner",
-    }
-    car = build_merchant_car(category, merchant, trigger)
-    evidence = extract_evidence(category, merchant, trigger, None, car)
-    assert fact_value_for_test(evidence, "retrieved_digest_title")
-    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
-    body = max(candidates, key=lambda c: c.total_score).body.lower()
-    assert "aligner" in body
-    assert "category match" in body or "practo" in body
-    assert "d_not_in_seed" not in body
-
-
-def test_cbr_similarity_adapts_familiar_trigger_with_shifted_merchant_state():
-    merchant = {**load_seed("merchants_seed.json", "merchants")[4], "offers": []}
-    category = load_category(merchant["category_slug"])
-    trigger = {
-        "id": "trg_review_theme_no_offer",
-        "scope": "merchant",
-        "kind": "review_theme_emerged",
-        "source": "review_digest",
-        "merchant_id": merchant["merchant_id"],
-        "payload": {
-            "theme": "delivery_late",
-            "occurrences_30d": 5,
-            "avg_delay_minutes": 39,
-            "common_quote": "late again",
-        },
-        "urgency": 3,
-        "suppression_key": "review:no-offer",
-    }
-    car = build_merchant_car(category, merchant, trigger)
-    match = retrieve_similar_case(car, trigger)
-    assert match and match[0] >= 0.60
-    assert case_similarity(car, trigger, match[1]) == match[0]
-
-    evidence = extract_evidence(category, merchant, trigger, None, car)
-    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
-    body = max(candidates, key=lambda c: c.total_score).body.lower()
-    assert "kitchen-dispatch problem" in body
-    assert "39" in body
-    assert "increase sales" not in body
-
-
-def test_property_based_compulsion_fires_without_exact_trigger_name():
-    merchant = load_seed("merchants_seed.json", "merchants")[6]
-    category = load_category(merchant["category_slug"])
-    trigger = {
-        "id": "trg_restart_window_custom",
-        "scope": "merchant",
-        "kind": "restart_window_custom",
-        "source": "ops_calendar",
-        "merchant_id": merchant["merchant_id"],
-        "payload": {
-            "available_slots": [{"label": "Mon 7 PM"}, {"label": "Wed 7 PM"}],
-            "active_members_at_risk": 28,
-            "metric": "member restarts",
-        },
-        "urgency": 3,
-        "suppression_key": "restart-window:custom",
-    }
-    car = build_merchant_car(category, merchant, trigger)
-    evidence = extract_evidence(category, merchant, trigger, None, car)
-    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
-    best = max(candidates, key=lambda c: c.total_score)
-    assert best.persuasion_principle == "scarcity"
-    assert best.map_scores["prompt"] >= 8
-    assert "mon 7 pm" in best.body.lower() or "wed 7 pm" in best.body.lower()
-
-
-def test_final_engagement_pass_adds_command_cta_line_break_and_peer_norm():
-    reset_state()
-    client = TestClient(app)
-    merchant = load_seed("merchants_seed.json", "merchants")[4]
-    trigger = {
-        "id": "trg_unseen_delivery_delay_engagement",
-        "scope": "merchant",
-        "kind": "delivery_delay_spike",
-        "source": "review_digest",
-        "merchant_id": merchant["merchant_id"],
-        "payload": {
-            "theme": "delivery_late",
-            "occurrences_30d": 6,
-            "avg_delay_minutes": 42,
-            "metric": "late delivery complaints",
-        },
-        "urgency": 3,
-        "suppression_key": "delivery-delay:engagement",
-    }
-    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
-    push(client, "merchant", merchant["merchant_id"], merchant)
-    push(client, "trigger", trigger["id"], trigger)
-    action = client.post("/v1/tick", json={"now": "2026-05-02T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
-    body = action["body"]
-    assert "\n\n" in body
-    assert "Want me" not in body
-    assert "Should I" not in body
-    assert "Reply YES" in body
-    assert "Restaurants in" in body
-
-
-def test_rationale_reads_like_decision_justification():
-    merchant = load_seed("merchants_seed.json", "merchants")[0]
-    category = load_category(merchant["category_slug"])
-    trigger = load_seed("triggers_seed.json", "triggers")[0]
-    car = build_merchant_car(category, merchant, trigger)
-    evidence = extract_evidence(category, merchant, trigger, None, car)
-    best = max(build_candidates(category, merchant, trigger, None, evidence, car), key=lambda c: c.total_score)
-    assert best.rationale.startswith("Trigger:")
-    assert "Frame:" in best.rationale
-    assert "Receptivity:" in best.rationale
-    assert "Suppression:" in best.rationale
-    assert "Selected signal" not in best.rationale
-
-
-def fact_value_for_test(evidence, label: str) -> str:
-    return next((item.value for item in evidence if item.label == label), "")
-
-
-def test_teardown_clears_state():
-    reset_state()
-    client = TestClient(app)
-    push(client, "category", "dentists", load_category("dentists"))
-    assert contexts
-    resp = client.post("/v1/teardown")
-    assert resp.json()["cleared"] is True
-    assert not contexts
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sys
+
+from fastapi.testclient import TestClient
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from app.decision_engine import build_candidates, build_merchant_car, case_similarity, constitutional_violations, extract_evidence, retrieve_similar_case, score_map, select_cialdini_principle  # noqa: E402
+from app.main import app, category_arm_pool, contexts, conversations, merchant_action_memory, merchant_auto_replies, merchant_opt_out, suppressed  # noqa: E402
+
+
+def load_seed(name: str, key: str) -> list[dict]:
+    return json.loads((ROOT / "dataset" / name).read_text(encoding="utf-8"))[key]
+
+
+def load_category(slug: str) -> dict:
+    return json.loads((ROOT / "dataset" / "categories" / f"{slug}.json").read_text(encoding="utf-8"))
+
+
+def reset_state() -> None:
+    contexts.clear()
+    conversations.clear()
+    suppressed.clear()
+    merchant_opt_out.clear()
+    merchant_auto_replies.clear()
+    merchant_action_memory.clear()
+    category_arm_pool.clear()
+
+
+def push(client: TestClient, scope: str, context_id: str, payload: dict, version: int = 1):
+    return client.post(
+        "/v1/context",
+        json={"scope": scope, "context_id": context_id, "version": version, "payload": payload, "delivered_at": "2026-04-26T10:00:00Z"},
+    )
+
+
+def test_health_and_context_idempotency():
+    reset_state()
+    client = TestClient(app)
+    assert client.get("/v1/healthz").json()["status"] == "ok"
+    cat = load_category("dentists")
+    resp = push(client, "category", "dentists", cat)
+    assert resp.status_code == 200
+    assert resp.json()["accepted"] is True
+    same = push(client, "category", "dentists", cat)
+    assert same.status_code == 200
+    assert same.json()["idempotent"] is True
+    stale = push(client, "category", "dentists", cat, version=0)
+    assert stale.status_code == 409
+
+
+def test_tick_creates_grounded_research_action():
+    reset_state()
+    client = TestClient(app)
+    merchants = load_seed("merchants_seed.json", "merchants")
+    triggers = load_seed("triggers_seed.json", "triggers")
+    merchant = merchants[0]
+    trigger = triggers[0]
+    push(client, "category", "dentists", load_category("dentists"))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "trigger", trigger["id"], trigger)
+
+    resp = client.post("/v1/tick", json={"now": "2026-04-26T10:30:00Z", "available_triggers": [trigger["id"]]})
+    assert resp.status_code == 200
+    actions = resp.json()["actions"]
+    assert len(actions) == 1
+    body = actions[0]["body"]
+    assert "JIDA" in body
+    assert "124" in body
+    assert actions[0]["send_as"] == "vera"
+    assert actions[0]["suppression_key"] == trigger["suppression_key"]
+
+
+def test_customer_consent_and_send_as():
+    reset_state()
+    client = TestClient(app)
+    merchants = load_seed("merchants_seed.json", "merchants")
+    customers = load_seed("customers_seed.json", "customers")
+    triggers = load_seed("triggers_seed.json", "triggers")
+    merchant = merchants[0]
+    customer = customers[0]
+    trigger = triggers[2]
+    push(client, "category", "dentists", load_category("dentists"))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "customer", customer["customer_id"], customer)
+    push(client, "trigger", trigger["id"], trigger)
+
+    resp = client.post("/v1/tick", json={"now": "2026-04-26T11:00:00Z", "available_triggers": [trigger["id"]]})
+    action = resp.json()["actions"][0]
+    assert action["send_as"] == "merchant_on_behalf"
+    assert action["customer_id"] == customer["customer_id"]
+    assert "Priya" in action["body"]
+    assert "Wed 5 Nov" in action["body"]
+
+
+def test_reply_replay_behaviors():
+    reset_state()
+    client = TestClient(app)
+    auto = "Thank you for contacting us! Our team will respond shortly."
+    first = client.post("/v1/reply", json={"conversation_id": "conv_auto", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 2}).json()
+    assert first["action"] == "wait"
+    second = client.post("/v1/reply", json={"conversation_id": "conv_auto", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 3}).json()
+    assert second["action"] == "wait"
+    third = client.post("/v1/reply", json={"conversation_id": "conv_auto", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 4}).json()
+    assert third["action"] == "end"
+
+    intent = client.post("/v1/reply", json={"conversation_id": "conv_intent", "merchant_id": "m1", "from_role": "merchant", "message": "Ok lets do it. Whats next?", "turn_number": 2}).json()
+    assert intent["action"] == "send"
+    assert "preparing" in intent["body"]
+    assert intent["cta"] == "none"
+
+    hostile = client.post("/v1/reply", json={"conversation_id": "conv_hostile", "merchant_id": "m1", "from_role": "merchant", "message": "Stop messaging me. This is useless spam.", "turn_number": 2}).json()
+    assert hostile["action"] == "end"
+
+    plain_stop = client.post("/v1/reply", json={"conversation_id": "conv_plain_stop", "merchant_id": "m1", "from_role": "merchant", "message": "STOP", "turn_number": 2}).json()
+    assert plain_stop["action"] == "end"
+
+    offtopic = client.post("/v1/reply", json={"conversation_id": "conv_offtopic", "merchant_id": "m1", "from_role": "merchant", "message": "What is the cricket score?", "turn_number": 2}).json()
+    assert offtopic["action"] == "send"
+    assert "outside" in offtopic["body"].lower()
+
+
+def test_ended_conversation_never_sends_again_and_omitted_merchant_optout():
+    reset_state()
+    client = TestClient(app)
+    conversations["conv_done"] = {"merchant_id": "m_done", "ended": True, "turns": []}
+    resp = client.post("/v1/reply", json={"conversation_id": "conv_done", "from_role": "merchant", "message": "hello", "turn_number": 5}).json()
+    assert resp["action"] == "end"
+
+    conversations["conv_stop"] = {"merchant_id": "m_stop", "ended": False, "turns": []}
+    stop = client.post("/v1/reply", json={"conversation_id": "conv_stop", "from_role": "merchant", "message": "Stop messaging me", "turn_number": 2}).json()
+    assert stop["action"] == "end"
+    assert "m_stop" in merchant_opt_out
+
+
+def test_validation_errors_are_challenge_style_400():
+    reset_state()
+    client = TestClient(app)
+    resp = client.post("/v1/context", json={"scope": "category"})
+    assert resp.status_code == 400
+    body = resp.json()
+    assert body["accepted"] is False
+    assert body["reason"] == "malformed"
+
+
+def test_merchant_level_auto_reply_tracking_across_conversations():
+    reset_state()
+    client = TestClient(app)
+    auto = "Thank you for contacting us! Our team will respond shortly."
+    assert client.post("/v1/reply", json={"conversation_id": "conv_auto_1", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 2}).json()["action"] == "wait"
+    assert client.post("/v1/reply", json={"conversation_id": "conv_auto_2", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 3}).json()["action"] == "wait"
+    assert client.post("/v1/reply", json={"conversation_id": "conv_auto_3", "merchant_id": "m1", "from_role": "merchant", "message": auto, "turn_number": 4}).json()["action"] == "end"
+
+
+def test_placeholder_triggers_do_not_leak_missing_fields():
+    reset_state()
+    client = TestClient(app)
+    merchant = load_seed("merchants_seed.json", "merchants")[5]
+    trigger = {
+        "id": "trg_placeholder_competitor",
+        "scope": "merchant",
+        "kind": "competitor_opened",
+        "source": "external",
+        "merchant_id": merchant["merchant_id"],
+        "customer_id": None,
+        "payload": {"placeholder": True, "metric_or_topic": "competitor_opened"},
+        "urgency": 2,
+        "suppression_key": "competitor:placeholder",
+        "expires_at": "2026-06-30T00:00:00Z",
+    }
+    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "trigger", trigger["id"], trigger)
+    action = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
+    assert "None" not in action["body"]
+    assert "the available context" not in action["body"]
+    assert action["cta"] == "binary_yes_no"
+
+
+def test_customer_without_matching_consent_routes_to_merchant_not_dead_line():
+    reset_state()
+    client = TestClient(app)
+    merchant = load_seed("merchants_seed.json", "merchants")[7]
+    customer = load_seed("customers_seed.json", "customers")[11]
+    trigger = {
+        "id": "trg_recall_no_scope",
+        "scope": "customer",
+        "kind": "recall_due",
+        "source": "internal",
+        "merchant_id": merchant["merchant_id"],
+        "customer_id": customer["customer_id"],
+        "payload": {"placeholder": True, "metric_or_topic": "recall_due"},
+        "urgency": 2,
+        "suppression_key": "recall:no_scope",
+        "expires_at": "2026-06-30T00:00:00Z",
+    }
+    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "customer", customer["customer_id"], customer)
+    push(client, "trigger", trigger["id"], trigger)
+    action = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
+    assert action["send_as"] == "vera"
+    assert action["cta"] == "binary_yes_no"
+    assert "I will not send" not in action["body"]
+
+
+def test_decision_engine_scores_case_anchor_shapes():
+    merchants = load_seed("merchants_seed.json", "merchants")
+    customers = load_seed("customers_seed.json", "customers")
+    triggers = load_seed("triggers_seed.json", "triggers")
+    cases = [
+        (merchants[0], triggers[0], None),
+        (merchants[0], triggers[2], customers[0]),
+        (merchants[4], triggers[9], None),
+        (merchants[6], triggers[13], None),
+        (merchants[8], triggers[17], None),
+        (merchants[8], triggers[18], customers[12]),
+    ]
+    for merchant, trigger, customer in cases:
+        category = load_category(merchant["category_slug"])
+        evidence = extract_evidence(category, merchant, trigger, customer)
+        candidates = build_candidates(category, merchant, trigger, customer, evidence)
+        assert candidates
+        best = max(candidates, key=lambda c: c.total_score)
+        assert best.total_score >= 36, (trigger["id"], best.total_score, best.body, best.rubric_scores)
+        assert any(ch.isdigit() for ch in best.body)
+        assert trigger["kind"].replace("_", " ") in best.body.lower() or any(e.source.startswith("trigger") for e in best.evidence)
+
+
+def test_full_expanded_dataset_proxy_has_no_weak_outputs():
+    import subprocess
+    import sys as _sys
+    subprocess.run([_sys.executable, "dataset/generate_dataset.py", "--seed-dir", "dataset", "--out", "expanded"], cwd=ROOT, check=True, capture_output=True)
+    result = subprocess.run([_sys.executable, "scripts/score_proxy.py", "34"], cwd=ROOT, text=True, capture_output=True)
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
+def test_car_map_jitai_and_best_of_n_debug_fields():
+    merchant = load_seed("merchants_seed.json", "merchants")[0]
+    trigger = load_seed("triggers_seed.json", "triggers")[0]
+    category = load_category(merchant["category_slug"])
+    car = build_merchant_car(category, merchant, trigger)
+    assert car.merchant_name != "unknown"
+    assert car.category == "dentists"
+    assert all(value is not None for value in car.summary().values())
+
+    evidence = extract_evidence(category, merchant, trigger, None, car)
+    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
+    assert len(candidates) >= 3
+    best = max(candidates, key=lambda c: c.total_score)
+    assert best.decision_plan if hasattr(best, "decision_plan") else True
+    assert best.car_summary["category"] == "dentists"
+    assert {"severity", "receptivity", "intervention_fit"} <= set(best.jitai_scores)
+    assert {"motivation", "ability", "prompt"} <= set(best.map_scores)
+    assert best.frame in {"loss_frame", "gain_frame", "certainty_frame", "social_proof", "professional_value", "effort_externalization"}
+
+
+def test_bmap_penalizes_high_friction_cta():
+    merchant = load_seed("merchants_seed.json", "merchants")[0]
+    trigger = load_seed("triggers_seed.json", "triggers")[0]
+    category = load_category(merchant["category_slug"])
+    car = build_merchant_car(category, merchant, trigger)
+    easy = score_map(car, trigger, "JIDA says 124 patients are relevant. Reply YES and I will draft it.", "binary_yes_no", "professional_value", [])
+    hard = score_map(car, trigger, "JIDA says 124 patients are relevant. Please call, log in, choose a campaign, upload a file, and configure delivery.", "open_ended", "professional_value", [])
+    assert easy["ability"] > hard["ability"]
+
+
+def test_frames_follow_trigger_shape_and_action_memory_changes_plan():
+    merchant = load_seed("merchants_seed.json", "merchants")[1]
+    category = load_category(merchant["category_slug"])
+    dip = load_seed("triggers_seed.json", "triggers")[3]
+    spike = {**dip, "id": "trg_spike_test", "kind": "perf_spike", "payload": {"metric": "calls", "delta_pct": 0.18}, "urgency": 2}
+    dip_candidates = build_candidates(category, merchant, dip, None, extract_evidence(category, merchant, dip))
+    spike_candidates = build_candidates(category, merchant, spike, None, extract_evidence(category, merchant, spike))
+    assert any(c.frame == "loss_frame" for c in dip_candidates)
+    assert any(c.frame == "gain_frame" for c in spike_candidates)
+
+    remembered = {**merchant, "__vera_memory": {"last_action_type": "recovery_nudge", "last_response_intent": "auto_reply", "repeated_action_count": 3, "no_reply_count": 2}}
+    car = build_merchant_car(category, remembered, dip)
+    evidence = extract_evidence(category, remembered, dip, None, car)
+    candidates = build_candidates(category, remembered, dip, None, evidence, car)
+    assert all(c.jitai_scores["receptivity"] <= 4 for c in candidates)
+
+
+def test_openrouter_calibration_skips_without_key(monkeypatch):
+    import subprocess
+    import sys as _sys
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    result = subprocess.run([_sys.executable, "scripts/geval_calibrate.py"], cwd=ROOT, text=True, capture_output=True)
+    assert result.returncode == 0
+    assert "skipped" in result.stdout.lower()
+
+
+def test_cialdini_constitution_and_tot_debug_fields():
+    merchant = load_seed("merchants_seed.json", "merchants")[0]
+    trigger = load_seed("triggers_seed.json", "triggers")[0]
+    category = load_category(merchant["category_slug"])
+    car = build_merchant_car(category, merchant, trigger)
+    evidence = extract_evidence(category, merchant, trigger, None, car)
+    principle = select_cialdini_principle(car, trigger, evidence, "professional_value")
+    assert principle in {"authority", "social_proof", "liking", "reciprocity", "scarcity", "commitment"}
+    bad = "Dear valued partner, want to increase sales? Contact us?"
+    violations = constitutional_violations(bad, car, trigger, "open_ended")
+    assert "generic_or_corporate_copy" in violations
+
+    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
+    best = max(candidates, key=lambda c: c.total_score)
+    assert best.thought_frames
+    assert best.persuasion_principle
+    assert best.reference_key.startswith("dentists:")
+    assert not best.constitutional_violations
+
+
+def test_category_empirical_prior_flows_into_car():
+    reset_state()
+    client = TestClient(app)
+    merchant = load_seed("merchants_seed.json", "merchants")[0]
+    trigger = load_seed("triggers_seed.json", "triggers")[0]
+    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "trigger", trigger["id"], trigger)
+    first = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
+    client.post("/v1/reply", json={"conversation_id": first["conversation_id"], "merchant_id": merchant["merchant_id"], "from_role": "merchant", "message": "yes go ahead", "turn_number": 2})
+    assert category_arm_pool[merchant["category_slug"]]
+    next_trigger = {**trigger, "id": "trg_research_next", "suppression_key": "research:next"}
+    push(client, "trigger", next_trigger["id"], next_trigger, version=1)
+    second = client.post("/v1/tick", json={"now": "2026-04-26T10:05:00Z", "available_triggers": [next_trigger["id"]]}).json()["actions"][0]
+    priors = second["decision_plan"]["car_summary"]["category_arm_priors"]
+    assert priors
+
+
+def test_sparse_context_fallback_stays_specific_and_safe():
+    category = load_category("restaurants")
+    merchant = {
+        "merchant_id": "m_sparse_restaurant",
+        "category_slug": "restaurants",
+        "identity": {"name": "Asha Cafe", "owner_first_name": "Asha", "locality": "Indiranagar"},
+        "performance": {},
+        "offers": [],
+        "customer_aggregate": {},
+        "signals": [],
+        "conversation_history": [],
+    }
+    trigger = {
+        "id": "trg_sparse_reactivation",
+        "scope": "merchant",
+        "kind": "merchant_inactive",
+        "source": "internal",
+        "merchant_id": merchant["merchant_id"],
+        "payload": {"days_inactive": 14},
+        "urgency": 1,
+        "expires_at": "2026-06-30T00:00:00Z",
+    }
+    car = build_merchant_car(category, merchant, trigger)
+    evidence = extract_evidence(category, merchant, trigger, None, car)
+    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
+    assert len(candidates) == 1
+    body = candidates[0].body
+    assert "Asha" in body
+    assert "Indiranagar" in body
+    assert "increase sales" not in body.lower()
+    assert "sparse_context_floor" in candidates[0].risk_flags
+
+
+def test_broad_suppression_keys_are_made_unique():
+    reset_state()
+    client = TestClient(app)
+    merchant = load_seed("merchants_seed.json", "merchants")[4]
+    category = load_category(merchant["category_slug"])
+    push(client, "category", merchant["category_slug"], category)
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    t1 = {"id": "trg_broad_1", "scope": "merchant", "kind": "curious_ask_due", "source": "internal", "merchant_id": merchant["merchant_id"], "payload": {"topic": "lunch"}, "urgency": 2, "suppression_key": "curious_ask_due"}
+    t2 = {"id": "trg_broad_2", "scope": "merchant", "kind": "curious_ask_due", "source": "internal", "merchant_id": merchant["merchant_id"], "payload": {"topic": "dinner"}, "urgency": 2, "suppression_key": "curious_ask_due"}
+    push(client, "trigger", t1["id"], t1)
+    push(client, "trigger", t2["id"], t2)
+    actions = client.post("/v1/tick", json={"now": "2026-05-02T10:00:00Z", "available_triggers": [t1["id"], t2["id"]]}).json()["actions"]
+    assert len(actions) == 2
+    assert actions[0]["suppression_key"] != actions[1]["suppression_key"]
+
+
+def test_context_updates_do_not_wipe_reply_memory():
+    reset_state()
+    client = TestClient(app)
+    merchant = load_seed("merchants_seed.json", "merchants")[0]
+    trigger = load_seed("triggers_seed.json", "triggers")[0]
+    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "trigger", trigger["id"], trigger)
+    first = client.post("/v1/tick", json={"now": "2026-04-26T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
+    client.post("/v1/reply", json={"conversation_id": first["conversation_id"], "merchant_id": merchant["merchant_id"], "from_role": "merchant", "message": "yes go ahead"})
+    push(client, "merchant", merchant["merchant_id"], {**merchant, "signals": ["fresh update"]}, version=2)
+    assert merchant_action_memory[merchant["merchant_id"]]["last_response_intent"] == "commitment"
+
+
+def test_pharmacy_without_consent_routes_to_merchant_and_avoids_medical_dispatch():
+    merchant = load_seed("merchants_seed.json", "merchants")[8]
+    customer = {**load_seed("customers_seed.json", "customers")[12], "consent": {"scope": []}}
+    trigger = load_seed("triggers_seed.json", "triggers")[18]
+    category = load_category(merchant["category_slug"])
+    car = build_merchant_car(category, merchant, trigger, customer)
+    evidence = extract_evidence(category, merchant, trigger, customer, car)
+    candidates = build_candidates(category, merchant, trigger, customer, evidence, car)
+    best = max(candidates, key=lambda c: c.total_score)
+    assert best.send_as == "vera"
+    assert "consent-safe" in best.body
+    assert "dispatch" not in best.body.lower()
+    assert "pharmacy_consent_or_medical_advice_risk" not in best.constitutional_violations
+
+
+def test_unseen_ops_trigger_uses_payload_without_known_template():
+    reset_state()
+    client = TestClient(app)
+    merchant = load_seed("merchants_seed.json", "merchants")[4]
+    trigger = {
+        "id": "trg_unseen_delivery_delay",
+        "scope": "merchant",
+        "kind": "delivery_delay_spike",
+        "source": "review_digest",
+        "merchant_id": merchant["merchant_id"],
+        "payload": {
+            "theme": "delivery_late",
+            "occurrences_30d": 6,
+            "avg_delay_minutes": 42,
+            "metric": "late delivery complaints",
+        },
+        "urgency": 3,
+        "suppression_key": "delivery-delay:unseen",
+        "expires_at": "2026-06-30T00:00:00Z",
+    }
+    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "trigger", trigger["id"], trigger)
+    action = client.post("/v1/tick", json={"now": "2026-05-02T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
+    body = action["body"].lower()
+    assert "42" in body
+    assert "delivery late" in body or "late delivery" in body
+    assert "prep/rider handoff" in body
+    assert "increase sales" not in body
+
+
+def test_missing_digest_id_retrieves_nearest_category_fact_for_research():
+    merchant = load_seed("merchants_seed.json", "merchants")[0]
+    category = load_category("dentists")
+    trigger = {
+        "id": "trg_unseen_aligner_research",
+        "scope": "merchant",
+        "kind": "research_digest",
+        "source": "external_digest",
+        "merchant_id": merchant["merchant_id"],
+        "payload": {
+            "top_item_id": "d_not_in_seed",
+            "topic": "clear aligner consultations",
+            "metric_or_topic": "clear aligners near me",
+        },
+        "urgency": 2,
+        "suppression_key": "research:unseen-aligner",
+    }
+    car = build_merchant_car(category, merchant, trigger)
+    evidence = extract_evidence(category, merchant, trigger, None, car)
+    assert fact_value_for_test(evidence, "retrieved_digest_title")
+    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
+    body = max(candidates, key=lambda c: c.total_score).body.lower()
+    assert "aligner" in body
+    assert "category match" in body or "practo" in body
+    assert "d_not_in_seed" not in body
+
+
+def test_cbr_similarity_adapts_familiar_trigger_with_shifted_merchant_state():
+    merchant = {**load_seed("merchants_seed.json", "merchants")[4], "offers": []}
+    category = load_category(merchant["category_slug"])
+    trigger = {
+        "id": "trg_review_theme_no_offer",
+        "scope": "merchant",
+        "kind": "review_theme_emerged",
+        "source": "review_digest",
+        "merchant_id": merchant["merchant_id"],
+        "payload": {
+            "theme": "delivery_late",
+            "occurrences_30d": 5,
+            "avg_delay_minutes": 39,
+            "common_quote": "late again",
+        },
+        "urgency": 3,
+        "suppression_key": "review:no-offer",
+    }
+    car = build_merchant_car(category, merchant, trigger)
+    match = retrieve_similar_case(car, trigger)
+    assert match and match[0] >= 0.60
+    assert case_similarity(car, trigger, match[1]) == match[0]
+
+    evidence = extract_evidence(category, merchant, trigger, None, car)
+    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
+    body = max(candidates, key=lambda c: c.total_score).body.lower()
+    assert "kitchen-dispatch problem" in body
+    assert "39" in body
+    assert "increase sales" not in body
+
+
+def test_property_based_compulsion_fires_without_exact_trigger_name():
+    merchant = load_seed("merchants_seed.json", "merchants")[6]
+    category = load_category(merchant["category_slug"])
+    trigger = {
+        "id": "trg_restart_window_custom",
+        "scope": "merchant",
+        "kind": "restart_window_custom",
+        "source": "ops_calendar",
+        "merchant_id": merchant["merchant_id"],
+        "payload": {
+            "available_slots": [{"label": "Mon 7 PM"}, {"label": "Wed 7 PM"}],
+            "active_members_at_risk": 28,
+            "metric": "member restarts",
+        },
+        "urgency": 3,
+        "suppression_key": "restart-window:custom",
+    }
+    car = build_merchant_car(category, merchant, trigger)
+    evidence = extract_evidence(category, merchant, trigger, None, car)
+    candidates = build_candidates(category, merchant, trigger, None, evidence, car)
+    best = max(candidates, key=lambda c: c.total_score)
+    assert best.persuasion_principle == "scarcity"
+    assert best.map_scores["prompt"] >= 8
+    assert "mon 7 pm" in best.body.lower() or "wed 7 pm" in best.body.lower()
+
+
+def test_final_engagement_pass_adds_command_cta_line_break_and_peer_norm():
+    reset_state()
+    client = TestClient(app)
+    merchant = load_seed("merchants_seed.json", "merchants")[4]
+    trigger = {
+        "id": "trg_unseen_delivery_delay_engagement",
+        "scope": "merchant",
+        "kind": "delivery_delay_spike",
+        "source": "review_digest",
+        "merchant_id": merchant["merchant_id"],
+        "payload": {
+            "theme": "delivery_late",
+            "occurrences_30d": 6,
+            "avg_delay_minutes": 42,
+            "metric": "late delivery complaints",
+        },
+        "urgency": 3,
+        "suppression_key": "delivery-delay:engagement",
+    }
+    push(client, "category", merchant["category_slug"], load_category(merchant["category_slug"]))
+    push(client, "merchant", merchant["merchant_id"], merchant)
+    push(client, "trigger", trigger["id"], trigger)
+    action = client.post("/v1/tick", json={"now": "2026-05-02T10:00:00Z", "available_triggers": [trigger["id"]]}).json()["actions"][0]
+    body = action["body"]
+    assert "\n\n" in body
+    assert "Want me" not in body
+    assert "Should I" not in body
+    assert "Reply YES" in body
+    assert "Restaurants in" in body
+
+
+def test_rationale_reads_like_decision_justification():
+    merchant = load_seed("merchants_seed.json", "merchants")[0]
+    category = load_category(merchant["category_slug"])
+    trigger = load_seed("triggers_seed.json", "triggers")[0]
+    car = build_merchant_car(category, merchant, trigger)
+    evidence = extract_evidence(category, merchant, trigger, None, car)
+    best = max(build_candidates(category, merchant, trigger, None, evidence, car), key=lambda c: c.total_score)
+    assert best.rationale.startswith("Trigger:")
+    assert "Frame:" in best.rationale
+    assert "Receptivity:" in best.rationale
+    assert "Suppression:" in best.rationale
+    assert "Selected signal" not in best.rationale
+
+
+def fact_value_for_test(evidence, label: str) -> str:
+    return next((item.value for item in evidence if item.label == label), "")
+
+
+def test_teardown_clears_state():
+    reset_state()
+    client = TestClient(app)
+    push(client, "category", "dentists", load_category("dentists"))
+    assert contexts
+    resp = client.post("/v1/teardown")
+    assert resp.json()["cleared"] is True
+    assert not contexts