Step 2 — Define proactive triggers

"""Proactive guidance state and trigger detection for the Inkeep onboarding bridge.

ProactiveState is kept separate from the context-pull code in copilot_server.py
so the two concerns don't interfere: pull_live_activity() answers "what has this
user been doing on Autoplay's connector" on demand; ProactiveState owns the
explicit demo-action counts and SSE fan-out for trigger delivery. Both are keyed
by the same user_id — there's no separate session concept to reconcile between them.
"""

from __future__ import annotations

import asyncio
import time
from collections import defaultdict, deque
from dataclasses import asdict, dataclass
from typing import Any
from uuid import uuid4

TRIGGER_WINDOW_SECONDS = 120
DATA_SOURCE_OPEN_THRESHOLD = 3


@dataclass
class GuidanceEvent:
    type: str          # "guidance_offer" or "guidance_start"
    user_id: str
    message: str
    target: str
    tour_id: str
    reason: str
    created_at: float


@dataclass
class ProactiveOffer:
    id: str
    user_id: str
    email: str | None
    message: str
    target: str
    tour_id: str
    reason: str
    created_at: float


class ProactiveState:
    """Lightweight in-memory store for the proactive guidance layer."""

    def __init__(self) -> None:
        self.recent_actions: dict[str, deque[dict[str, Any]]] = defaultdict(
            lambda: deque(maxlen=30)
        )
        self.pending_offer_by_user: dict[str, ProactiveOffer] = {}
        self.guidance_subscribers: dict[str, set[asyncio.Queue[GuidanceEvent]]] = defaultdict(set)

    def store_actions(self, user_id: str, actions: list[dict[str, Any]]) -> None:
        bucket = self.recent_actions[user_id]
        now = time.time()
        for action in actions:
            item = dict(action)
            item.setdefault("timestamp_start", now)
            bucket.append(item)

    async def publish_guidance(self, event: GuidanceEvent) -> None:
        for queue in list(self.guidance_subscribers[event.user_id]):
            await queue.put(event)


def _is_data_source_open(action: dict[str, Any]) -> bool:
    return action.get("type", "") == "data_source_open"


def _detect_stuck_onboarding(
    *, user_id: str, email: str | None, state: ProactiveState
) -> ProactiveOffer | None:
    # Guard: fire at most once per user
    if user_id in state.pending_offer_by_user:
        return None

    cutoff = time.time() - TRIGGER_WINDOW_SECONDS
    recent = [
        a for a in state.recent_actions[user_id]
        if float(a.get("timestamp_start") or 0) >= cutoff
    ]

    # If the user already completed the step in this window, don't nudge them
    completed = any(
        a.get("type") == "data_source_complete" for a in recent
    )
    if completed:
        return None

    open_clicks = [a for a in recent if _is_data_source_open(a)]
    if len(open_clicks) < DATA_SOURCE_OPEN_THRESHOLD:
        return None

    return ProactiveOffer(
        id=f"offer_{uuid4().hex}",
        user_id=user_id,
        email=email,
        message=(
            "Having trouble connecting your data source? "
            "The most common issue is IP whitelisting — "
            "want me to walk you through it?"
        ),
        target="[data-guide='connect-data-source']",
        tour_id="data-source-connection",
        reason=f"{len(open_clicks)} data_source_open actions in {TRIGGER_WINDOW_SECONDS}s without completion",
        created_at=time.time(),
    )

@app.get("/guidance/stream/{user_id}")
async def guidance_stream(user_id: str, request: Request) -> StreamingResponse:
    queue: asyncio.Queue[GuidanceEvent] = asyncio.Queue()
    proactive_state.guidance_subscribers[user_id].add(queue)

    async def events():
        try:
            # Send a ready event immediately so the client knows it's connected
            yield "event: ready\ndata: {}\n\n"
            while not await request.is_disconnected():
                try:
                    event = await asyncio.wait_for(queue.get(), timeout=15)
                    yield f"event: {event.type}\ndata: {json.dumps(asdict(event))}\n\n"
                except asyncio.TimeoutError:
                    yield "event: heartbeat\ndata: {}\n\n"
        finally:
            proactive_state.guidance_subscribers[user_id].discard(queue)

    return StreamingResponse(events(), media_type="text/event-stream")


@app.get("/context/{user_id}")
async def get_context(user_id: str) -> dict[str, Any]:
    """Return assembled context for injection into InkeepEmbeddedChat.

    Merges two sources:
    1. A fresh REST pull of Autoplay live activity (PostHog autocapture events via Step 1).
    2. Demo-specific stuck-connection narrative (explicit data_source_open events).

    Replaces the Step 1 version of this endpoint with one that is also
    aware of the proactive guidance state.
    """
    # Source 1 — Autoplay live activity, pulled fresh on every request
    live_activity = await pull_live_activity(user_id)

    # Source 2 — explicit demo actions tracked by ProactiveState
    demo_actions = list(proactive_state.recent_actions.get(user_id, []))

    if not live_activity and not demo_actions:
        return {"context": "", "has_activity": False, "user_id": user_id}

    lines = []
    for a in demo_actions[-10:]:  # last 10 explicit events
        atype = a.get("type", "")
        title = a.get("title", "")
        ts_raw = a.get("timestamp_start", 0)
        try:
            ts = float(ts_raw)
        except (TypeError, ValueError):
            ts = 0.0
        lines.append(f"- [{atype}] {title} (t={ts:.0f})")

    parts = []
    if live_activity:
        parts.append(live_activity)
    if lines:
        parts.append(
            "The user is working through the Nexus Cloud onboarding. "
            "Here is what they have been doing recently:\n"
            + "\n".join(lines)
            + "\n\nThey appear to be struggling with the Connect Data Source step. "
            "Start by acknowledging this and offer specific guidance."
        )

    context = "\n\n".join(parts)
    return {
        "context": context,
        "has_activity": bool(context.strip()),
        "user_id": user_id,
    }


@app.post("/inkeep/reply")
async def inkeep_reply(reply: InkeepReply) -> dict[str, Any]:
    offer = proactive_state.pending_offer_by_user.pop(reply.user_id, None)
    if offer is None:
        raise HTTPException(
            status_code=404, detail="No pending offer for this user"
        )
    if reply.offer_id and offer.id != reply.offer_id:
        raise HTTPException(status_code=409, detail="Offer ID does not match")

    accepted = GuidanceEvent(
        type="guidance_start",
        user_id=offer.user_id,
        message="Start the data source connection guide.",
        target=offer.target,
        tour_id=offer.tour_id,
        reason=offer.reason,
        created_at=time.time(),
    )
    await proactive_state.publish_guidance(accepted)
    return {"accepted": True, "guidance": asdict(accepted)}


@app.post("/demo/actions")
async def demo_actions(payload: DemoActionsPayload) -> dict[str, Any]:
    await handle_demo_actions(payload)
    offer = proactive_state.pending_offer_by_user.get(payload.user_id)
    return {
        "stored": len(payload.actions),
        "pending_offer": offer is not None,
        "offer": asdict(offer) if offer else None,
    }

const API_BASE = process.env.NEXT_PUBLIC_BRIDGE_URL ?? "http://localhost:8787";

export async function sendDataSourceOpen(userId: string): Promise<void> {
  await fetch(`${API_BASE}/demo/actions`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      user_id: userId,
      actions: [
        {
          type: "data_source_open",
          title: "Connect Data Source — opened",
          canonical_url: "https://app.example.com/onboarding",
        },
      ],
    }),
  });
}

export function guidanceStreamUrl(userId: string): string {
  return `${API_BASE}/guidance/stream/${encodeURIComponent(userId)}`;
}

export async function fetchContext(userId: string): Promise<string> {
  const response = await fetch(
    `${API_BASE}/context/${encodeURIComponent(userId)}`
  );
  if (!response.ok) return "";
  const data = await response.json();
  return (data as { context?: string }).context ?? "";
}

export async function acceptInkeepOffer(
  userId: string,
  offerId: string | undefined,
): Promise<void> {
  await fetch(`${API_BASE}/inkeep/reply`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ user_id: userId, offer_id: offerId, text: "yes" }),
  });
}

.proactivePopup {
  position: fixed;
  bottom: 84px;
  right: 24px;
  width: 320px;
  background: var(--panel);
  border: 1px solid var(--border);
  border-radius: 12px;
  box-shadow: 0 8px 32px rgba(0,0,0,.14);
  padding: 14px 16px;
  z-index: 300;
  animation: popIn 180ms ease;
}

@keyframes popIn {
  from { opacity: 0; transform: translateY(10px); }
  to   { opacity: 1; transform: translateY(0); }
}

.proactivePopup__x {
  position: absolute;
  top: 10px;
  right: 12px;
  background: none;
  border: none;
  cursor: pointer;
  font-size: 18px;
  line-height: 1;
  color: var(--muted);
}
.proactivePopup__x:hover { color: var(--ink); }

.proactivePopup__header {
  display: flex;
  align-items: center;
  gap: 7px;
  font-size: 11px;
  font-weight: 700;
  letter-spacing: 0.06em;
  color: var(--primary);
  text-transform: uppercase;
  margin-bottom: 8px;
}
.proactivePopup__icon {
  width: 22px;
  height: 22px;
  border-radius: 50%;
  background: var(--primary);
  color: #fff;
  display: flex;
  align-items: center;
  justify-content: center;
  flex-shrink: 0;
}
.proactivePopup__msg {
  font-size: 13.5px;
  line-height: 1.5;
  margin-bottom: 12px;
  padding-right: 20px;
}
.proactivePopup__actions {
  display: flex;
  flex-direction: column;
  gap: 8px;
}