Day 2 — Worktree B — Escalation Taxonomy Split

You are a Sonnet subagent in an isolated worktree. You have NO conversation history. This document is your full brief. Read it through, plan, then execute.

This worktree is life-safety scope. A regression here means a real caller in crisis gets routed to a sleepy director instead of 988. The contract test is mandatory.

Run model + setup

Model: Sonnet (you).
Subagent type: voice-agent-engineer.

Working dir: temp worktree. From there:

git fetch origin
git checkout -b feat/verticals-platform-day2-B-escalation-split origin/feat/verticals-platform-day1-foundation

DO NOT push to main or feat/verticals-platform-day1-foundation. Open a PR back to feat/verticals-platform-day1-foundation.
DO NOT merge your own PR.

Read first (in order)

C:/dev/knowledge/specs/day2-verticals-platform/00-MAIN-ORCHESTRATOR.md — orchestrator handoff.
C:/Users/johnm/.claude/plans/steady-munching-penguin.md — read sections "Layer 4 — Escalation taxonomy" and "Phase 3 (escalation split)" verification.
C:/dev/knowledge/voice-clients/README.md — the two-track distinction is documented here.
C:/dev/voice-agent-livekit/safety.py — current monolithic safety handler.
C:/dev/voice-agent-livekit/moderation.py — pattern matching (THREAT_PATTERNS, CRISIS_PATTERNS, DV).
C:/dev/voice-agent-livekit/core/notifications.py — notification routing (send_threat_alert_, send_crisis_alert_, send_urgent_callback_*).
C:/dev/voice-agent-livekit/verticals/church/tools.py line 73+ — current _request_callback (Track A).
C:/dev/src/lib/moderation.ts — chatbot moderation (cooldown/temp_block/permanent_block tree).
C:/dev/src/app/api/chatbot/stream/route.ts lines 86–336 — chatbot crisis-detection branch (CRISIS_PATTERNS regex around line 100, override at 289).

The new crisis_events table (created Day 1; verify schema):

# Quick schema check via supabase mcp:
SELECT column_name, data_type FROM information_schema.columns WHERE table_name='crisis_events' ORDER BY ordinal_position;

The two-track contract (the rule you're enforcing in code)

	Track A: Operational handoff	Track B: Safety/Crisis escalation
Examples	"Can I speak with a director?" / At-need death / Demand-for-human / Urgent legal	Suicide ideation / DV / Threat / Abuse
Tool	`transfer_to_director` (live SIP, Worktree A) OR `request_callback` (async)	`flag_safety_event` (NEW, this worktree)
Live bridge?	YES	NO — defer to 988/911/professional hotline
Routes to	`notification_phone` + `notification_email`	`support@churchwiseai.com` ALWAYS + optional `crisis_contact_phone` (separate from `notification_phone`)
DB row	`voice_callback_requests`	NEW `crisis_events` (created Day 1)
`moderation_violations`	NOT used	Still written (existing — for chatbot too)
Notification template	Operational urgent	Crisis-specific (different copy, different recipient list)

The bug we are fixing: today both tracks share request_callback, share notification_phone, and conflate at routing time. A frustrated caller demanding a director ("urgent" on Track A) and a suicide ideation event ("pastoral_emergency" on Track A) both go to the same SMS to the same director. This worktree separates them.

Scope

B.1 — Create `voice-agent-livekit/core/escalation.py`

Two top-level async functions. Both are vertical-agnostic and read TenantConfig.

"""core/escalation.py — Two-track escalation routing.

Track A (operational handoff): caller asks for human, AI judges this needs a
human. Routes to notification_phone + notification_email. May SIP-bridge if
configured. Writes voice_callback_requests.

Track B (safety/crisis escalation): suicide / DV / threat / abuse / AI-detected
out-of-scope. Routes to support@ ALWAYS + optional crisis_contact_phone.
NEVER live-bridges. Writes crisis_events.

THIS MODULE IS LIFE-SAFETY. CODEOWNERS-gated. Contract-tested. Do not modify
without contract test passing on the change.
"""
from __future__ import annotations

from typing import Any, Literal
from loguru import logger

from core.tenant_config import TenantConfig

OperationalUrgency = Literal["normal", "urgent", "at_need", "pastoral_emergency"]
SafetyEventType = Literal["crisis", "threat", "abuse", "dv"]
SafetyAction = Literal["hangup", "stay_with_resources", "inject_988", "inject_dv_hotline", "log_only"]


async def handle_operational_handoff(
    *,
    config: TenantConfig,
    supabase: Any,
    caller_name: str,
    caller_phone: str,
    reason: str,
    urgency: OperationalUrgency = "normal",
    preferred_time: str | None = None,
    agreed_day: str | None = None,
    agreed_time_window: str | None = None,
    source: Literal["voice", "chat"] = "voice",
) -> dict:
    """Track A. Writes voice_callback_requests. Notifies notification_*.

    Returns: {success, callback_id, message, sms_sent, email_sent}.
    """
    # 1. INSERT into voice_callback_requests
    # 2. If urgency in (urgent, at_need, pastoral_emergency) → SMS to config.notification_phone via core/notifications.send_urgent_callback_sms
    # 3. Always send email to config.notification_email via core/notifications.send_callback_request_email
    # 4. Return success + callback_id (UUID) + spoken message ("I'll have {director} call you {response_time_promise_text}.")
    ...


async def handle_safety_event(
    *,
    config: TenantConfig,
    supabase: Any,
    event_type: SafetyEventType,
    source: Literal["voice", "chat"],
    session_id: str | None,
    caller_phone: str | None,
    caller_identifier: str | None,
    detected_pattern: str,
    detected_text_redacted: str,
    action_taken: SafetyAction,
    resources_recited: list[str] | None = None,
    conversation_continued: bool = False,
) -> dict:
    """Track B. Writes crisis_events. Notifies support@ ALWAYS + optional crisis_contact_phone SMS.

    NEVER attempts a SIP bridge — even if config.operational_handoff.transfer_enabled is true.
    Crisis = professional hotline + post-event notification, never a live human bridge.

    Returns: {success, crisis_event_id, support_notified, crisis_contact_notified}.
    """
    # 1. INSERT into crisis_events with all fields
    # 2. Always email support@churchwiseai.com via core/notifications.send_crisis_alert_to_support
    # 3. If config.safety.crisis_contact_phone is set AND event_type != 'threat-only' → SMS via core/notifications.send_crisis_contact_sms (NEW helper)
    # 4. NEVER call lk_api.SIP.create_sip_participant from this function. The crisis gate in core/transfer.py is the second-line defense; this function is the first.
    # 5. Update crisis_events row with notification status timestamps
    # 6. Return success + crisis_event_id (UUID) + status
    ...

Implement both fully. Use core.notifications helpers; add new ones if needed (e.g. send_crisis_contact_sms distinct from send_urgent_callback_sms).

B.2 — Add `flag_safety_event` tool to the agent

Register on the shared CoordinatorAgent (currently verticals/church/agents.py, soon core/_shared_agents.py if Worktree A pulls it). Tool signature:

@function_tool()
async def flag_safety_event(
    self,
    context: RunContext,
    event_type: Literal["crisis", "threat", "abuse", "dv"],
    detected_text: str,
):
    """Mark a safety event for escalation. Track B.

    LIFE-SAFETY: Use this when a caller expresses suicide ideation, threats,
    abuse, or domestic violence. Never use request_callback or
    transfer_to_director for these. The system enforces this — but you should
    also ALWAYS recite the relevant hotline (988, DV, 911) before or after
    firing this tool.
    """
    # delegates to core.escalation.handle_safety_event
    ...

Update voice-agent-livekit/verticals/church/prompts.py and voice-agent-livekit/verticals/funeral/prompts.py to teach the agent: "When a safety pattern fires (988, DV, threat), say the hotline first, then call flag_safety_event with the right event_type." Forbid the agent from using request_callback for crisis.

B.3 — Migrate `safety.py` and `moderation.py` to route via the split

In voice-agent-livekit/safety.py, the existing check_threat/check_crisis/check_dv/check_abuse handlers currently call _notify_threat, _notify_crisis, etc. directly. Replace each with a call to core.escalation.handle_safety_event() with the appropriate event_type and action_taken.

Preserve current behaviors:

THREAT → action_taken='hangup', _hangup_after_playout('threat') still fires
CRISIS → action_taken='inject_988', conversation_continued=True, agent stays on call
DV → action_taken='inject_dv_hotline', conversation_continued=True
ABUSE 2nd offense → action_taken='hangup', hangup still fires

moderation.py continues to write moderation_violations (it's the chatbot's restriction policy log; we don't remove that). But ALSO the safety.py path now writes crisis_events via the new handle_safety_event for the AUDIT trail.

B.4 — Mirror the split in the chatbot

In src/app/api/chatbot/stream/route.ts:

The existing CRISIS_PATTERNS regex check (around line 289) currently calls logViolation() which writes to moderation_violations. Add a parallel write to crisis_events:

// After CRISIS_PATTERNS match:
await logViolation({ /* existing */ });
await logCrisisEvent({
  tenant_id: churchId,
  vertical: vertical,
  event_type: 'crisis',
  source: 'chat',
  session_id: conversationId,
  caller_identifier: visitorIdentifier,
  detected_pattern: matchedPattern,
  detected_text_redacted: redactPii(message).slice(0, 200),
  action_taken: 'inject_988',
  resources_recited: ['988', 'us_dv_hotline'],
  conversation_continued: true,
});
// Also notify support@:
await sendCrisisAlertToSupport({ tenant_id: churchId, ... });

Add helper logCrisisEvent to src/lib/moderation.ts or a new src/lib/crisis-events.ts (your call — match existing patterns). Use service role Supabase client.

If tenant_voice_agents.crisis_contact_phone is set, send SMS to it via existing Telnyx/Twilio helper.

B.5 — Notification template split

In voice-agent-livekit/core/notifications.py:

Confirm the existing send_crisis_alert_to_support, send_threat_alert_* helpers still exist and work (Day 1 didn't change them).
ADD new helper send_crisis_contact_sms(phone, *, tenant_name, event_type, detected_text_redacted) for the optional crisis_contact_phone SMS path. Distinct from send_urgent_callback_sms (which is operational urgency).
Crisis email template: explicit warning that this is a SAFETY event, NOT an operational callback. Includes 988 / DV hotline / 911 links. Subject prefix: [SAFETY EVENT — review immediately].
Operational urgent email template: subject prefix [Urgent callback]. NO 988 references (avoid alarming callers' families).

B.6 — The 100-message contract test (mandatory)

Create voice-agent-livekit/tests/test_escalation_routing.py. Use pytest-asyncio.

The test feeds 100 simulator messages (50 operational, 50 safety/crisis) and asserts each goes to the correct table + notification path.

OPERATIONAL_PHRASES = [
    "Can I speak with a director?",
    "I want to talk to a real person",
    "Get me your manager",
    "Can the pastor call me back?",
    "I need to schedule a meeting with the funeral director",
    # ... 50 total, mix of "demand human", "urgent question", "at-need"
]

CRISIS_PHRASES = [
    "I want to end my life",
    "I'm thinking about suicide",
    "I'm going to kill myself",
    "no one would miss me",
    "He's been hitting me",
    "She threatened to kill the kids",
    # ... 50 total, covering crisis + threat + DV + abuse
]


@pytest.mark.parametrize("phrase", OPERATIONAL_PHRASES)
async def test_operational_routes_to_callback(phrase):
    # Feed phrase into simulator session
    result = await simulate(phrase)
    # Assert voice_callback_requests row written
    assert "voice_callback_requests" in result.tables_written
    # Assert crisis_events NOT written
    assert "crisis_events" not in result.tables_written
    # Assert notification went to notification_phone (NOT crisis_contact_phone)
    assert result.sms_destination == "notification_phone"
    assert result.email_to == "notification_email"


@pytest.mark.parametrize("phrase", CRISIS_PHRASES)
async def test_crisis_routes_to_safety_event(phrase):
    result = await simulate(phrase)
    assert "crisis_events" in result.tables_written
    assert "voice_callback_requests" not in result.tables_written
    assert result.email_to == "support@churchwiseai.com"
    # If crisis_contact_phone is set, SMS routes there (NOT to notification_phone)
    # If crisis_contact_phone is NOT set, NO SMS at all
    assert result.sms_destination != "notification_phone"

Implement simulate() as a thin test harness that mocks Supabase + notification helpers and captures which functions are called. Don't depend on a live LLM — use a stubbed agent that just routes based on regex/keywords (the production agent uses LLM judgment but the contract test runs against the deterministic fast-path).

The contract test must pass before PR is merged.

B.7 — CODEOWNERS gate

Create or update .github/CODEOWNERS:

# Life-safety: changes require founder review
voice-agent-livekit/core/escalation.py @JohnMoelker
voice-agent-livekit/safety.py @JohnMoelker
voice-agent-livekit/moderation.py @JohnMoelker
voice-agent-livekit/tests/test_escalation_routing.py @JohnMoelker
voice-agent-livekit/tests/test_transfer_crisis_gate.py @JohnMoelker
voice-agent-livekit/core/transfer.py @JohnMoelker
src/lib/moderation.ts @JohnMoelker
src/lib/crisis-events.ts @JohnMoelker
src/app/api/chatbot/stream/route.ts @JohnMoelker

B.8 — Update vertical prompts

In voice-agent-livekit/verticals/funeral/prompts.py and voice-agent-livekit/verticals/church/prompts.py, add explicit instructions:

SAFETY ESCALATION RULES (life-safety, never violate):
If caller mentions suicide, self-harm, ending their life: recite 988 immediately, then call flag_safety_event(event_type='crisis').
If caller mentions domestic violence, abuse, being hit: recite National DV Hotline (1-800-799-7233), then call flag_safety_event(event_type='dv').
If caller threatens violence to others: recite 911, then call flag_safety_event(event_type='threat').
NEVER call request_callback for safety events. NEVER call transfer_to_director for safety events.
Crisis events stay on the line WITH the caller (you continue the conversation). Threats end the call after the resource is recited.

Files you will modify

voice-agent-livekit/safety.py                                   # route via handle_safety_event
voice-agent-livekit/moderation.py                               # extend to write crisis_events
voice-agent-livekit/core/notifications.py                       # add send_crisis_contact_sms
voice-agent-livekit/verticals/church/prompts.py                 # safety escalation rules
voice-agent-livekit/verticals/funeral/prompts.py                # safety escalation rules
voice-agent-livekit/verticals/church/agents.py                  # register flag_safety_event tool
src/lib/moderation.ts                                            # add logCrisisEvent OR import from new module
src/app/api/chatbot/stream/route.ts                              # add crisis_events write parallel to moderation_violations

Files you will create

voice-agent-livekit/core/escalation.py
voice-agent-livekit/tests/test_escalation_routing.py
src/lib/crisis-events.ts                                          # OR add to existing moderation.ts; your call
.github/CODEOWNERS                                                # if not exists; otherwise edit

Files NOT to touch

voice-agent-livekit/core/transfer.py — Worktree A owns this. Coordinate via shared imports.
voice-agent-livekit/core/tenant_config.py — Day 1 stable. Don't add fields without coordination.
voice-agent-livekit/verticals/church/agents.py:transfer_to_director — Worktree A is updating this.
src/app/api/chatbot/stream/route.ts:request_callback tool — Worktree A is adding this.
src/middleware.ts — Worktree C owns this.
src/lib/verticals/** — Worktree C creates this.
src/app/founder/[token]/voice-clients/** — Worktree D creates this.
crisis_events TABLE — already created Day 1, just write to it.

Verification (run before opening PR)

cd voice-agent-livekit
python -m py_compile core/escalation.py safety.py moderation.py verticals/church/prompts.py verticals/funeral/prompts.py
python -m pytest tests/test_escalation_routing.py -v        # 100 cases must pass
python -m pytest tests/                                      # full suite — ensure no regression

cd ..
npx tsc --noEmit --skipLibCheck src/app/api/chatbot/stream/route.ts src/lib/moderation.ts src/lib/crisis-events.ts

# Quick schema check (via supabase mcp_execute_sql, READ-ONLY):
# SELECT column_name FROM information_schema.columns WHERE table_name='crisis_events';
# Confirm: id, tenant_id, vertical, event_type, severity, source, action_taken, etc. exist

When done

git add -A
git commit -m "feat(verticals): Day 2 Worktree B — escalation taxonomy split (Track A vs Track B)"
git push -u origin feat/verticals-platform-day2-B-escalation-split
gh pr create \
  --base feat/verticals-platform-day1-foundation \
  --title "feat(verticals): Day 2 B — escalation split — operational vs safety/crisis routing" \
  --body "$(cat <<'EOF'
## Summary
Day 2 Worktree B. Splits the conflated `request_callback` path into two tracks. Adds `core/escalation.py` with `handle_operational_handoff()` (Track A) and `handle_safety_event()` (Track B). Adds `flag_safety_event` agent tool. Routes safety events to `crisis_events` (NEW Day 1 table) + `support@` ALWAYS + optional `crisis_contact_phone`. NEVER live-bridges crisis. Mirrors the split in chatbot at `/api/chatbot/stream`. Adds 100-message contract test. Adds CODEOWNERS gate on life-safety files.

See `knowledge/specs/day2-verticals-platform/02-WORKTREE-B-escalation-split.md` for the full spec.

## Test plan
- [ ] `python -m pytest voice-agent-livekit/tests/test_escalation_routing.py -v` — 50 operational + 50 crisis cases route to correct table + correct destination
- [ ] `python -m pytest voice-agent-livekit/tests/` — no regressions in existing suite
- [ ] `npx tsc --noEmit src/app/api/chatbot/stream/route.ts src/lib/moderation.ts` — type-check
- [ ] Manual (Day 3): say "I want to end my life" to chatbot → crisis_events row + support@ email + NO SMS to notification_phone
- [ ] Manual (Day 3): say "Can I speak with the pastor?" to chatbot → voice_callback_requests row + SMS to notification_phone

## Migration applied
- None (uses Day 1 `crisis_events` table)

🤖 Generated with [Claude Code](https://claude.com/claude-code)
EOF
)"

Then in your final message back to the orchestrator (≤300 words):

PR URL
Commit SHA(s)
Verification status
Any deviations from this spec + why
Anything that needs to be in Day 3 integration checklist
Confirmation that the contract test passes

What "done" looks like

core/escalation.py exists with both functions implemented end-to-end (not stubs).
flag_safety_event tool registered on the church Coordinator (which serves all verticals via instructions_override).
safety.py routes via handle_safety_event (preserves all behaviors).
Chatbot writes crisis_events parallel to moderation_violations on crisis detection.
Contract test passes (100 cases).
CODEOWNERS gate in place.
One PR open against feat/verticals-platform-day1-foundation. NOT MERGED.

Hard rules (do not violate)

NEVER live-bridge a crisis event. The first line of defense is handle_safety_event (won't even call SIP). The second line is the crisis gate inside core/transfer.py (Worktree A owns; you don't change it).
NEVER notify the operational notification_phone for crisis events.
NEVER skip the contract test. If it fails, fix the code, not the test.
NEVER push to main or to feat/verticals-platform-day1-foundation.

You're cleared to start.

Run model + setup​

Read first (in order)​

The two-track contract (the rule you're enforcing in code)​

Scope​

B.1 — Create voice-agent-livekit/core/escalation.py​

B.2 — Add flag_safety_event tool to the agent​

B.3 — Migrate safety.py and moderation.py to route via the split​

B.4 — Mirror the split in the chatbot​

B.5 — Notification template split​

B.6 — The 100-message contract test (mandatory)​

B.7 — CODEOWNERS gate​

B.8 — Update vertical prompts​

Files you will modify​

Files you will create​

Files NOT to touch​

Verification (run before opening PR)​

When done​

What "done" looks like​

Hard rules (do not violate)​