Step 1 — Connect real-time events

# server.py — Autoplay → Dify real-time knowledge server (session-scoped)
import asyncio, os
from collections import defaultdict
from datetime import datetime, timezone
from typing import Optional

import openai
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from autoplay_sdk import AsyncConnectorClient, AsyncSessionSummarizer
from autoplay_sdk.agent_context import AsyncAgentContextWriter

# ── Config ───────────────────────────────────────────────────────────
CONNECTOR_URL = os.environ["AUTOPLAY_STREAM_URL"]   # e.g. https://…/stream/<product_id>
API_TOKEN     = os.environ["AUTOPLAY_API_TOKEN"]
MAX_CHUNKS    = 50   # per session — keeps memory bounded

app = FastAPI()
async_openai = openai.AsyncOpenAI()

# ── Request model ─────────────────────────────────────────────────────
# Dify always calls POST /retrieval with this JSON body.
# knowledge_id is the External Knowledge ID you set in the Dify UI —
# it contains whichever session identifier (session_id or email) your
# frontend passed via the chat-messages `inputs` field.
class RetrievalSetting(BaseModel):
    top_k: int = 5
    score_threshold: float = 0.5

class RetrievalRequest(BaseModel):
    knowledge_id: str
    query: str
    retrieval_setting: RetrievalSetting

# ── In-memory stores (session-scoped) ────────────────────────────────
chunks_by_session: dict[str, list[dict]] = defaultdict(list)
email_to_session: dict[str, str] = {}


def _store(session_id: str, email: Optional[str], text: str) -> None:
    entry = {"text": text, "ts": datetime.now(timezone.utc).isoformat()}
    buf = chunks_by_session[session_id]
    buf.append(entry)
    if len(buf) > MAX_CHUNKS:
        buf.pop(0)
    if email:
        email_to_session[email] = session_id


# ── Callbacks ────────────────────────────────────────────────────────
async def write_actions(session_id: str, text: str) -> None:
    _store(session_id, email=None, text=text)


async def on_raw_payload(payload: dict) -> None:
    """Capture email → session_id mapping from every incoming event."""
    sid   = payload.get("session_id")
    email = payload.get("email")
    if sid and email:
        email_to_session[email] = sid


async def overwrite_with_summary(session_id: str, summary: str) -> None:
    chunks_by_session[session_id] = [{
        "text": summary,
        "ts": datetime.now(timezone.utc).isoformat(),
    }]


# ── LLM summarizer ───────────────────────────────────────────────────
async def llm(prompt: str) -> str:
    r = await async_openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3, max_tokens=256,
    )
    return r.choices[0].message.content


summarizer   = AsyncSessionSummarizer(llm=llm, threshold=20)
agent_writer = AsyncAgentContextWriter(
    summarizer=summarizer,
    write_actions=write_actions,
    overwrite_with_summary=overwrite_with_summary,
    debounce_ms=0,
)


# ── POST /retrieval ───────────────────────────────────────────────────
# Dify appends /retrieval to the base URL you registered and sends a
# POST with knowledge_id, query, and retrieval_setting in the body.
# It never sends GET requests or query parameters to this endpoint.
@app.post("/retrieval")
async def retrieval(body: RetrievalRequest):
    """
    knowledge_id contains the session identifier your frontend passed
    via Dify's `inputs` field when opening the conversation.

    Accepts either:
      knowledge_id = "ps_abc123"         — direct session_id lookup
      knowledge_id = "user@example.com"  — resolved via email index

    Response contract Dify expects:
      { "records": [{"content": "…", "score": 1.0, "title": "…", "metadata": {}}, …] }

    Note: metadata must be {} (empty object) — never null.
    """
    identifier = body.knowledge_id

    # Try direct session_id lookup first
    source = chunks_by_session.get(identifier)

    # Fall back to email → session_id index
    if source is None:
        sid = email_to_session.get(identifier)
        source = chunks_by_session.get(sid) if sid else None

    if not source:
        # No events received yet — return empty gracefully
        return JSONResponse({"records": []})

    top_k     = body.retrieval_setting.top_k
    threshold = body.retrieval_setting.score_threshold

    records = [
        {
            "content": c["text"],
            "score": 1.0,
            "title": f"events @ {c['ts']}",
            "metadata": {},          # must be {} — never null
        }
        for c in reversed(source)    # newest first
        if 1.0 >= threshold
    ][:top_k]

    return JSONResponse({"records": records})


# ── Background stream task ────────────────────────────────────────────
@app.on_event("startup")
async def start_stream():
    asyncio.create_task(_run_stream())


async def _run_stream():
    async with AsyncConnectorClient(url=CONNECTOR_URL, token=API_TOKEN) as client:
        client.on_actions(agent_writer.add)
        client.on_raw(on_raw_payload)
        await client.run()

You are a friendly and helpful assistant for users of this product.

Focus on helping people find their way in the UI, complete workflows, and
understand features. Assume some users are seeing the product for the first time.

## 💬 How to use the "Current User Activity" record

You may receive a special record titled "Current User Activity" in the
retrieved context. This shows what THIS user has been doing on the
platform in the last 2 minutes — which page they are on and what they
clicked. The activity is scoped to their session, so it reflects only
their actions, not anyone else's.

{{#context#}}

When this record is present:

1. **Acknowledge their activity naturally** — for example:
   "I can see you're currently on the Projects page" or
   "It looks like you've been exploring the Dashboard."

2. **Use it to give specific directions** — instead of generic
   instructions, reference where they are:
   "From the page you're on, click the blue 'Add Project' button
   at the top right."

3. **Detect if they might be lost** — if their actions show them
   clicking around without a clear pattern, gently offer help:
   "It looks like you might be looking for something specific.
   Can I help you find it?"

4. **Don't force it** — if the user's question has nothing to do
   with their current activity, just answer the question normally.
   Don't mention their activity unless it's helpful.

## ❓ How to answer questions

- **Be specific**: reference actual button names, tab labels, and
  menu items from the knowledge base.
- **Use numbered steps**: when explaining how to do something,
  always use a numbered list.
- **Keep it simple**: avoid technical jargon. Explain as if the
  user has never used the platform before.
- **Be encouraging**: use phrases like "Great question!" or
  "That's easy to do" to make users feel comfortable.
- **Offer next steps**: after answering, suggest what they might
  want to do next.
- **Admit when you don't know**: if the knowledge base doesn't
  have the answer, say so honestly.

## 🌐 Language

Respond in the same language the user writes in.

## ✅ Examples of good responses

User is on the Dashboard, asks "How do I create a project?":
  "I can see you're currently on the Dashboard. To create a new
   project:
   1. Click on 'My Projects' in the left sidebar
   2. Click the 'Add Project' button at the top right
   3. Choose the type, template, or options that match what you're creating
   4. Fill in the required details and click 'Create'
   Would you like me to explain what each field means?"

User is on the Invoice page, asks "Where are settings?":
  "The settings aren't on this page — you can find them by clicking
   on your profile icon in the top right corner, then selecting
   'Settings' from the dropdown menu."

User has no activity context, asks "What can I do here?":
  "Welcome! Here's what you can do:
   1. Dashboard — see an overview of your work
   2. My Projects — create and manage projects
   3. Reports — view analytics or exports
   4. Billing — manage invoices or account settings
   What would you like to explore first?"