You are a senior Python/FastAPI engineer working inside my existing Replit project for LIA (Loyalty Intelligence Agent). Your job is to implement the conversation state machine and logging using the schema that already exists in Supabase.

IMPORTANT: 
- Do NOT change database structure.
- Do NOT drop or alter existing columns.
- Only use the columns that already exist.
- I cannot code; you must find the right files and make the edits.

────────────────────────────────
STACK / DATA MODEL CONTEXT

Backend:
- Python + FastAPI
- Async DB (Supabase / Postgres), likely via asyncpg or Supabase client

Key tables (ALREADY CREATED):

1) dp_conversations
- conversation_id uuid NOT NULL DEFAULT gen_random_uuid()
- customer_id     uuid NOT NULL
- dealer_id       uuid NOT NULL
- template_key    text
- stage           text
- status          text DEFAULT 'active'
- created_at      timestamptz DEFAULT now()
- updated_at      timestamptz DEFAULT now()
- defer_until     date
- defer_reason    text
- from_number     text
- to_number       text
- channel         text
- campaign        text
- soft_no_count   integer DEFAULT 0
- appointment_status text DEFAULT 'NONE'
- escalation_reason text
- escalation_at   timestamptz
- confusion_count integer NOT NULL DEFAULT 0   -- added
- needs_human_review boolean NOT NULL DEFAULT FALSE -- added

Constraints (you can assume):
- status must be one of: 'active', 'paused', 'dnc'
- appointment_status must be one of: 'NONE', 'INVITED', 'BOOKED'

2) dp_messages
- message_id     uuid NOT NULL DEFAULT gen_random_uuid()
- conversation_id uuid
- direction      text NOT NULL           -- inbound | outbound
- channel        text NOT NULL
- sender         text NOT NULL           -- lia | customer | manager
- message_body   text NOT NULL
- created_at     timestamptz DEFAULT now()  -- this is the message timestamp
- template_key   text
- mode           text                     -- campaign | intro | nudge | tier_reply | appointment
- updated_at     timestamp DEFAULT now()
- ab_variant     text

There may already be a PK on message_id and FK to dp_conversations; assume basic integrity is in place.

────────────────────────────────
BUSINESS RULES TO IMPLEMENT

TASK 2B – Conversation Status Flow

Use dp_conversations.status with three values:
- 'active'
- 'paused'
- 'dnc'

Rules:

1) active → paused WHEN:
   - soft_no_count >= 2
   OR
   - confusion_count >= 2
   OR
   - conversation is escalated (via an /escalate endpoint)

2) active → dnc WHEN:
   - opt-out is detected by the URE (tier = 'T6')

3) paused → active WHEN:
   - a manager sends a reply via the dashboard AND
   - the escalation has been resolved (we will treat the manager reply as the resolution)

4) dnc → no further messages allowed, except a one-off confirmation if we want.

TASK 2C – Appointment Status Flow

Use dp_conversations.appointment_status with three values:
- 'NONE'
- 'INVITED'
- 'BOOKED'

Rules:

1) NONE → INVITED WHEN:
   - URE tier = 'T3_APPOINTMENT'
   AND
   - the engine classifies the customer message as a vague appointment intent
     (e.g. "Saturday morning", "sometime next week", "yeah interested")

2) INVITED → BOOKED WHEN:
   - the engine detects a specific time (e.g. "9:30 on Saturday", "11am Tuesday 14th")

3) BOOKED → stays BOOKED unless a brand new conversation row is created.
   (Do NOT reset BOOKED in this logic.)

Assume the URE or engine already provides:
- tier: string, e.g. "T3_APPOINTMENT", "T6", etc.
- appointment_intent: one of "none", "vague", "specific"

TASK 2D – Manager Override Behaviour

When a manager replies from the dashboard for a given conversation_id:

- Set status = 'active'
- Set needs_human_review = FALSE
- Set escalation_reason = NULL
- Do NOT change appointment_status unless explicitly provided
- Log the manager message into dp_messages with sender='manager'

TASK 2E – dp_messages Logging

Every message (customer, LIA, manager) must log a row in dp_messages:

- message_body      (text)
- sender            ('lia' | 'customer' | 'manager')
- direction         ('inbound' | 'outbound')
- channel           (e.g. 'whatsapp')
- template_key      (nullable, if applicable)
- mode              ('campaign' | 'intro' | 'nudge' | 'tier_reply' | 'appointment')
- created_at        (default now(), leave to DB unless overriding)
- ab_variant        (optional, e.g. 'A' / 'B' for experiments)

We already have these columns; you only need to implement a reusable logging helper and call it in the right places.

────────────────────────────────
WHAT YOU NEED TO BUILD / MODIFY

You must locate the following (or create them cleanly if missing):

1) A conversation state helper module  
   Suggested file: `app/services/conversation_state.py` (or similar)  

   Implement a pure function that takes the current row values and returns the new status:

   - Input:
     - current_status: 'active' | 'paused' | 'dnc'
     - tier: string | None
     - soft_no_count: int
     - confusion_count: int
     - needs_human_review: bool

   - Output:
     - new_status: 'active' | 'paused' | 'dnc'

   Logic:
   - If current_status == 'dnc' → always return 'dnc' (terminal)
   - Else if tier == 'T6' → return 'dnc'
   - Else if needs_human_review is TRUE OR soft_no_count >= 2 OR confusion_count >= 2 → return 'paused'
   - Else → return 'active'

   Then create an async function that:
   - Loads the dp_conversations row by conversation_id
   - Computes the new status using the function above
   - If it has changed, updates status and updated_at.

2) Appointment status helper module  
   Suggested file: `app/services/appointment_state.py`

   Implement a pure function:

   - Input:
     - current_status: 'NONE' | 'INVITED' | 'BOOKED' (default to 'NONE' if NULL)
     - tier: string | None
     - appointment_intent: 'none' | 'vague' | 'specific'

   - Output:
     - new_appointment_status

   Logic:
   - If current_status == 'BOOKED' → return 'BOOKED'
   - Else if appointment_intent == 'specific' → return 'BOOKED'
   - Else if current_status == 'NONE' AND appointment_intent == 'vague' AND tier == 'T3_APPOINTMENT' → return 'INVITED'
   - Else → return current_status

   And an async function that:
   - Loads the current appointment_status for the conversation
   - Applies the function above
   - Updates appointment_status and updated_at if it changed.

3) Escalation endpoint  
   Suggested file: `app/routes/conversations.py` or similar

   Implement:

   - POST `/api/conversations/{conversation_id}/escalate`
   - Request body: `{ "escalation_reason": "string" }`
   - Behaviour:
     - Update dp_conversations for that conversation_id:
       - needs_human_review = TRUE
       - escalation_reason = provided reason
       - escalation_at = now()
       - status = 'paused'
       - updated_at = now()

4) Manager reply endpoint  

   Implement:

   - POST `/api/conversations/{conversation_id}/manager_reply`
   - Request body includes:
     - message_body: string
     - template_key: optional string
     - mode: optional string (default to 'campaign' if not provided)
     - channel: optional string (default 'whatsapp' if not provided)

   Behaviour:
   - Ensure the conversation exists
   - Update dp_conversations:
     - status = 'active'
     - needs_human_review = FALSE
     - escalation_reason = NULL
     - updated_at = now()
   - Log a dp_messages row using the shared logging helper (see point 5):
     - sender = 'manager'
     - direction = 'outbound'
     - channel = given / default
     - message_body, template_key, mode from payload

5) Shared message logging helper  

   Suggested file: `app/services/message_log.py` (or similar)

   Implement an async function:

   ```python
   async def log_message(
       conversation_id: str,
       message_body: str,
       sender: str,      # 'lia' | 'customer' | 'manager'
       direction: str,   # 'inbound' | 'outbound'
       channel: str,
       template_key: str | None,
       mode: str | None,
       ab_variant: str | None = None,
   ) -> None:
       ...