Py.Cafe

# app.py — Solara therapeutic chatbot with advanced silence/typing handling

"""
A Solara web‑app that streams GPT‑4 responses and reacts compassionately to
silence or long typing pauses, following Dutch trauma‑informed guidelines.
"""

from __future__ import annotations

import asyncio
import os
import random
from datetime import datetime
from typing import List, cast

from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageParam
from typing_extensions import TypedDict

import solara
import solara.lab

# ------------------------------------------------------------------
# UI helpers
# ------------------------------------------------------------------

# Monkey‑patch Solara's default placeholder text to Dutch
_original_TextField = solara.v.TextField

def _custom_TextField(*args, **kwargs):
    if kwargs.get("label") == "Type a message...":
        kwargs["label"] = "Typ een bericht..."
    return _original_TextField(*args, **kwargs)

solara.v.TextField = _custom_TextField

# ------------------------------------------------------------------
# Reactive state
# ------------------------------------------------------------------

class MessageDict(TypedDict):
    role: str  # "user" | "assistant" | "system"
    content: str

messages: solara.Reactive[List[MessageDict]] = solara.reactive([])
input_text: solara.Reactive[str] = solara.reactive("")

# ------------------------------------------------------------------
# OpenAI setup
# ------------------------------------------------------------------
try:
    import pycafe  # type: ignore

    OPENAI_API_KEY = pycafe.get_secret(
        "OPENAI_API_KEY",
        """You need an OpenAI API key to run the chatbot.\n""",
    )
except ModuleNotFoundError:
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

openai = AsyncOpenAI(api_key=OPENAI_API_KEY) if OPENAI_API_KEY else None

# ------------------------------------------------------------------
# Constants & timer‑related globals
# ------------------------------------------------------------------
APP_START_TIME = datetime.now()

inactivity_task: asyncio.Task | None = None  # 30‑60 s silence
typing_task: asyncio.Task | None = None      # 2‑3 min typing pause
followup_task: asyncio.Task | None = None    # post‑invitation scenarios

# ------------------------------------------------------------------
# Helper functions
# ------------------------------------------------------------------

def next_silence_threshold() -> int:  # 30‑60 s
    return random.randint(30, 60)

def next_typing_threshold() -> int:  # 2‑3 min
    return random.randint(120, 180)

def nl_time_system_message() -> MessageDict:
    """Return current local timestamp as system message."""
    return {
        "role": "system",
        "content": f"Current NL time: {datetime.now():%Y-%m-%d %H:%M:%S}",
    }

def no_api_key_message():
    messages.value = [
        {
            "role": "assistant",
            "content": "No OpenAI API key found. Please set OPENAI_API_KEY.",
        }
    ]

# ------------------------------------------------------------------
# Streaming wrapper
# ------------------------------------------------------------------

def _append_system_and_stream(prompt: str):
    """Append a system prompt then stream GPT‑4 reply into messages."""
    messages.value = [*messages.value, {"role": "system", "content": prompt}]
    if openai is None:
        no_api_key_message()
        return

    async def _stream():
        response = await openai.chat.completions.create(
            model="gpt-4-1106-preview",
            messages=cast(List[ChatCompletionMessageParam], messages.value),
            stream=True,
            temperature=0.5,
        )
        messages.value = [*messages.value, {"role": "assistant", "content": ""}]
        async for chunk in response:
            if chunk.choices[0].finish_reason == "stop":
                break
            delta = chunk.choices[0].delta.content or ""
            messages.value[-1]["content"] += delta
    asyncio.create_task(_stream())

# ------------------------------------------------------------------
# Silence / typing invitation timers
# ------------------------------------------------------------------

def restart_inactivity_timer():
    """(Re)start the 30‑60 s silence detector."""
    global inactivity_task
    if inactivity_task and not inactivity_task.done():
        inactivity_task.cancel()

    T = next_silence_threshold()
    user_msgs_before = sum(1 for m in messages.value if m["role"] == "user")

    async def _watch():
        await asyncio.sleep(T)
        # if user spoke meanwhile, abort
        if any(m["role"] == "user" for m in messages.value[user_msgs_before:]):
            return
        typed_since = bool(input_text.value.strip())

        # ---------- build invitation prompt -----------------------
        if user_msgs_before == 0 and not typed_since:  # opening silence
            invite = (
                f"Note to the therapeutic assistant: about {T} seconds of silence "
                "have passed since the session start and the client hasn't "
                "typed. Craft a gentle Dutch opener: validate that starting can "
                "feel awkward, silence is welcome, you're here whenever they're "
                "ready, optionally add a light personal aside. Keep it warm." 
            )
        elif user_msgs_before == 0 and typed_since:  # opening typing pause
            invite = (
                f"Note to the therapeutic assistant: the client has been typing "
                f"for about {T} seconds without sending their first message. "
                "Write a Dutch reply acknowledging it takes courage to press "
                "send, reassure there's no rush, optionally a light personal "
                "aside."
            )
        elif user_msgs_before > 0 and not typed_since:  # mid‑conversation silence
            invite = (
                f"Note to the therapeutic assistant: about {T} seconds of silence "
                "since the client's last message, no typing activity. "
                "Reply in Dutch: briefly reference their last share, validate the "
                "pause, invite noticing body sensations and which system might "
                "be active; vary wording if similar pauses happened before." 
            )
        else:  # mid‑conversation typing pause
            invite = (
                f"Note to the therapeutic assistant: the client has been typing "
                f"for about {T} seconds without sending. "
                "Reply in Dutch: note it's fine to take time, reflect last topic, "
                "offer simple way to answer (bullet points, feeling word), keep "
                "tone supportive; vary wording if repeat event."
            )

        # send invitation
        _append_system_and_stream(invite)
        # schedule deeper follow‑ups
        schedule_no_reply_followups("silence", user_msgs_before)

    inactivity_task = asyncio.create_task(_watch())


def restart_typing_timer():
    """(Re)start the 2‑3 min long‑typing detector."""
    global typing_task
    if typing_task and not typing_task.done():
        typing_task.cancel()

    T = next_typing_threshold()
    user_msgs_before = sum(1 for m in messages.value if m["role"] == "user")

    async def _watch():
        await asyncio.sleep(T)
        if not input_text.value.strip():
            return  # user cleared field
        # no new sent message?
        if any(m["role"] == "user" for m in messages.value[user_msgs_before:]):
            return

        # This counts as a typing‑pause invitation, so reuse inactivity logic
        if user_msgs_before == 0:
            invite = (
                f"Note to the therapeutic assistant: the client has been typing "
                f"for about {T} seconds without sending the first message. "
                "Dutch reply: acknowledge courage, reassure unlimited time, "
                "optional light aside."
            )
        else:
            invite = (
                f"Note to the therapeutic assistant: about {T} seconds of typing "
                "without sending, mid‑conversation. Dutch reply: it's okay to "
                "take time, reflect last topic, suggest easier response path." 
            )

        _append_system_and_stream(invite)
        schedule_no_reply_followups("typing", user_msgs_before)

    typing_task = asyncio.create_task(_watch())

# ------------------------------------------------------------------
# Follow‑up chain after invitations
# ------------------------------------------------------------------

def schedule_no_reply_followups(invitation_kind: str, user_msgs_before: int):
    """Handle 3‑4 min silence after an invitation and subsequent wrap‑up."""
    global followup_task
    if followup_task and not followup_task.done():
        followup_task.cancel()

    T_first = random.randint(180, 240)  # 3‑4 min

    async def _follow():
        await asyncio.sleep(T_first)
        # Did user speak meanwhile?
        if any(m["role"] == "user" for m in messages.value[user_msgs_before:]):
            return
        typed_since_invite = bool(input_text.value.strip())

        # ---------------- choose scenario prompt -------------------
        if user_msgs_before == 0 and not typed_since_invite:  # 1A
            prompt = (
                "Therapeutic system note · Scenario 1A\n"
                "Context: ~3‑4 min of full silence since invitation; no typing.\n\n"
                "Respond in Dutch: acknowledge the quiet as okay and even "
                "beautiful; reassure you're here whenever needed; optionally "
                "share a brief casual aside (e.g. 'ik schenk nog wat thee'). "
                "After this stay dormant until client responds."
            )
            _append_system_and_stream(prompt)
            return

        if user_msgs_before == 0 and typed_since_invite:  # 1B
            prompt = (
                "Therapeutic system note · Scenario 1B\n"
                "Context: Client typed but never sent first message (~3‑4 min).\n\n"
                "Dutch reply: validate that finding words takes courage, there's "
                "all the time in the world; optional light aside; remain dormant "
                "afterwards until client responds."
            )
            _append_system_and_stream(prompt)
            return

        # mid‑conversation variants
        if user_msgs_before > 0 and not typed_since_invite:  # 2A
            prompt = (
                "Therapeutic system note · Scenario 2A\n"
                "Context: Previous messages exist; client silent (~3‑4 min), "
                "no typing.\n\n"
                "Dutch: briefly reference last topic, validate pause, invite body "
                "signals & system check, optional brief aside. If still no reply "
                "after 1‑2 min ask about closing with summary; if still silent "
                "produce summary & ask for edits."
            )
            _append_system_and_stream(prompt)
            await _wrap_up_sequence(user_msgs_before)
            return

        # 2B
        prompt = (
            "Therapeutic system note · Scenario 2B\n"
            "Context: Previous messages exist; client typing for ~3‑4 min "
            "without sending.\n\n"
            "Dutch: note it's fine to take time, reflect last topic, suggest an "
            "easier way to respond; optional small suggestion/change of topic. "
            "Same wrap‑up flow as 2A if silence continues."
        )
        _append_system_and_stream(prompt)
        await _wrap_up_sequence(user_msgs_before)

    followup_task = asyncio.create_task(_follow())


async def _wrap_up_sequence(user_msgs_before: int):
    """Runs the 1‑2 min wrap‑up → summary chain for 2A/2B."""
    T_wrap = random.randint(60, 120)

    # ask about wrap‑up
    await asyncio.sleep(T_wrap)
    if any(m["role"] == "user" for m in messages.value[user_msgs_before:]):
        return
    wrap_prompt = (
        "Therapeutic system note · wrap‑up request\n"
        "No reply after gentle check‑in. Ask in Dutch, softly, if they'd like "
        "to finish with a short summary for their therapist (no extra details)."
    )
    _append_system_and_stream(wrap_prompt)

    # if still silent, send summary
    await asyncio.sleep(T_wrap)
    if any(m["role"] == "user" for m in messages.value[user_msgs_before:]):
        return
    summary_prompt = (
        "Therapeutic system note · produce summary\n"
        "Still no reply. Provide a concise Dutch summary (key feelings, "
        "interventions discussed, next steps), then ask if they want to add or "
        "change anything."
    )
    _append_system_and_stream(summary_prompt)

# ------------------------------------------------------------------
# System instructions & initialisation
# ------------------------------------------------------------------

SYSTEM_INSTRUCTIONS = """<the long metaprompt from the original code>"""
# ensure first system message is instructions
if not messages.value or messages.value[0]["role"] != "system":
    messages.value.insert(0, {"role": "system", "content": SYSTEM_INSTRUCTIONS})
elif messages.value[0]["content"] != SYSTEM_INSTRUCTIONS:
    messages.value[0] = {"role": "system", "content": SYSTEM_INSTRUCTIONS}

# ------------------------------------------------------------------
# Main send callback
# ------------------------------------------------------------------

@solara.lab.task
async def promt_ai(message: str):
    if openai is None:
        no_api_key_message()
        return

    # cancel ongoing timers because user spoke
    for t in (inactivity_task, typing_task, followup_task):
        if t and not t.done():
            t.cancel()

    # clear typing field
    input_text.set("")

    # add user message plus timestamp
    messages.value = [
        *messages.value,
        nl_time_system_message(),
        {"role": "user", "content": message},
    ]

    # stream assistant reply
    response = await openai.chat.completions.create(
        model="gpt-4-1106-preview",
        messages=cast(List[ChatCompletionMessageParam], messages.value),
        stream=True,
        temperature=0.5,
    )
    messages.value = [*messages.value, {"role": "assistant", "content": ""}]

    async for chunk in response:
        if chunk.choices[0].finish_reason == "stop":
            break
        delta = chunk.choices[0].delta.content or ""
        messages.value[-1]["content"] += delta

    # restart silence detector after assistant finishes
    restart_inactivity_timer()

# ------------------------------------------------------------------
# Solara page
# ------------------------------------------------------------------

@solara.component
def Page():
    with solara.Column(style={"width": "100%", "height": "50vh"}):
        with solara.lab.ChatBox():
            for item in messages.value:
                if item["role"] == "system":
                    continue
                with solara.lab.ChatMessage(
                    user=item["role"] == "user",
                    avatar=False,
                    name="Kompassie" if item["role"] == "assistant" else "Gebruiker",
                    color="rgba(0,0,0,0.06)" if item["role"] == "assistant" else "#ff991f",
                    avatar_background_color="primary" if item["role"] == "assistant" else None,
                    border_radius="20px",
                ):
                    solara.Markdown(item["content"])
        if promt_ai.pending:
            solara.Text("Ik denk erover na...", style={"font-size": "1rem", "padding-left": "20px"})
            solara.ProgressLinear()

        # ChatInput bound to reactive input_text
        solara.lab.ChatInput(
            value=input_text,
            on_value=lambda v: (
                input_text.set(v),
                restart_typing_timer() if v.strip() else None
            ),
            send_callback=promt_ai,
            disabled_send=promt_ai.pending,
            autofocus=True,
        ).key("input")

# schedule first inactivity timer when app starts
restart_inactivity_timer()