Step 2 — Define proactive triggers

"""Proactive triggers for the Rasa copilot.

Single custom trigger: `bulk_edit_opportunity` — fires when the user is
editing list items one-by-one and the SDK detects the pattern. The bot then
offers to walk the user through the (hidden) bulk-edit feature.

This is the SDK's actual value-prop in miniature: a multi-signal pattern
across pages and time (3+ Save-changes clicks within a short window) that
no single page or `setTimeout` could catch.

Delivery contract: the caller passes an async
`deliver(user_id, ProactiveTriggerResult)` coroutine that the
driver invokes when the trigger fires AND the FSM allows it. There is no
`session_id` in this contract — the REST live-activity read has no session
concept, so everything here is scoped by `user_id` alone.
"""

from __future__ import annotations

import asyncio
import logging
from collections.abc import Awaitable, Callable
from datetime import datetime, timedelta, timezone
from typing import Iterable

from autoplay_sdk.agent_state.v2.states import AgentStateV2, SessionState
from autoplay_sdk.proactive.triggers import (
    PredicateProactiveTrigger,
    ProactiveTriggerContext,
    ProactiveTriggerRegistry,
    ProactiveTriggerResult,
)

log = logging.getLogger("copilot.proactive")

TRIGGER_ID_BULK_EDIT_OPPORTUNITY = "bulk_edit_opportunity"

# Substring matched against `action.canonical_url` (case-insensitive). Swap
# for your own list page.
_PROJECTS_URL_FRAGMENT = "/projects"

# Substring matched against `action.title` — the human-readable label PostHog
# autocapture computes from the clicked element's text content.
_SAVE_CLICK_HINT = "save changes"
_BULK_TOGGLE_HINT = "bulk edit"

_MIN_SAVES = 3
_WINDOW_SECONDS = 120.0


def _is_projects_page(action) -> bool:
    return _PROJECTS_URL_FRAGMENT in (action.canonical_url or "").lower()


def _is_save_changes_click(action) -> bool:
    return _SAVE_CLICK_HINT in (action.title or "").lower()


def _is_bulk_toggle_click(action) -> bool:
    """If the user has already discovered bulk-edit in this session, don't
    nudge them about it."""
    title = (action.title or "").lower()
    return _BULK_TOGGLE_HINT in title or "apply to" in title


def _action_start_dt(action) -> datetime | None:
    """SlimAction.timestamp_start → tz-aware datetime.

    The SDK ships timestamps in two flavors depending on the source:
      - UNIX epoch as a numeric string ("1779188762.919")  ← PostHog destination
      - ISO-8601 ("2026-05-19T11:07:19.847Z")              ← other sources
    Handle both.
    """
    raw = getattr(action, "timestamp_start", None)
    if raw is None or raw == "":
        return None
    if isinstance(raw, datetime):
        return raw if raw.tzinfo else raw.replace(tzinfo=timezone.utc)
    s = str(raw).strip()
    try:
        return datetime.fromtimestamp(float(s), tz=timezone.utc)
    except (TypeError, ValueError):
        pass
    try:
        dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
        return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
    except (TypeError, ValueError):
        return None


def _latest_action_by_timestamp(actions):
    """Return the action with the largest timestamp_start, or None.
    PostHog batches don't always arrive in chronological order — `actions[-1]`
    is NOT reliably the most recent action."""
    latest, latest_ts = None, float("-inf")
    for a in actions:
        try:
            ts = float(a.timestamp_start) if a.timestamp_start else float("-inf")
        except (TypeError, ValueError):
            ts = float("-inf")
        if ts > latest_ts:
            latest_ts, latest = ts, a
    return latest


def _bulk_edit_opportunity_predicate(ctx: ProactiveTriggerContext) -> bool:
    """Fire when the user is editing list items one-by-one.

    Logic:
      1. The user must currently be on the list page (so the toast/tour land
         in context). If they wandered off, no nudge.
      2. The user must NOT have already discovered the bulk feature in the
         recent window.
      3. There must be ≥ _MIN_SAVES "Save changes" clicks on the list page
         within the trailing _WINDOW_SECONDS.
    """
    actions = ctx.recent_actions
    if not actions:
        return False

    latest = _latest_action_by_timestamp(actions)
    if latest is None or not _is_projects_page(latest):
        return False

    if any(_is_bulk_toggle_click(a) for a in actions):
        return False

    cutoff = datetime.now(timezone.utc) - timedelta(seconds=_WINDOW_SECONDS)
    save_count = 0
    for a in actions:
        if not _is_projects_page(a):
            continue
        if not _is_save_changes_click(a):
            continue
        ts = _action_start_dt(a)
        if ts is None or ts < cutoff:
            continue
        save_count += 1
        if save_count >= _MIN_SAVES:
            return True
    return False


def build_registry() -> ProactiveTriggerRegistry:
    """One custom trigger. The SDK ships built-ins (`canonical_url_ping_pong`,
    `user_page_dwell`, `section_playbook_match`) you can compose alongside —
    see the [authoring guide](/sdk/proactive-triggers-authoring) for examples."""
    bulk_edit = PredicateProactiveTrigger(
        trigger_id=TRIGGER_ID_BULK_EDIT_OPPORTUNITY,
        body=(
            "Updating projects one by one? There's a multi-select bulk edit "
            "you might not have seen — want me to show you?"
        ),
        predicate=_bulk_edit_opportunity_predicate,
    )
    return ProactiveTriggerRegistry([bulk_edit])

# ---------------------------------------------------------------------------
# Tick driver
# ---------------------------------------------------------------------------

DeliverFn = Callable[[str, ProactiveTriggerResult], Awaitable[None]]
UserContextFn = Callable[[str], Awaitable[ProactiveTriggerContext | None]]
UserStateFn = Callable[[str], SessionState]
ActiveUsersFn = Callable[[], Iterable[str]]


class ProactiveDriver:
    """Periodic tick loop that walks currently-active users and fires triggers."""

    def __init__(
        self,
        *,
        registry: ProactiveTriggerRegistry,
        get_active_users: ActiveUsersFn,
        get_context: UserContextFn,
        get_state: UserStateFn,
        deliver: DeliverFn,
        poll_seconds: float = 10.0,
    ) -> None:
        self._registry = registry
        self._get_active_users = get_active_users
        self._get_context = get_context
        self._get_state = get_state
        self._deliver = deliver
        self._poll_seconds = poll_seconds
        self._task: asyncio.Task | None = None
        self._running = False

    async def _tick_once(self) -> None:
        for user_id in list(self._get_active_users()):
            try:
                state = self._get_state(user_id)
                state.tick()  # auto-exit PROACTIVE/REACTIVE on timeout

                # This is the one line that changed from a local-buffer read
                # to a network call — everything below it is untouched.
                ctx = await self._get_context(user_id)
                if ctx is None:
                    continue

                result = self._registry.evaluate_first(ctx)
                if result is None:
                    continue

                # FSM gate. `transition_to_proactive` raises
                # InvalidTransitionError from non-THINKING states.
                if state.current_state != AgentStateV2.THINKING:
                    continue
                if not state.transition_to_proactive(result.trigger_id):
                    continue  # cooldown blocked it

                log.info(
                    "proactive: firing user=%s trigger=%s",
                    user_id, result.trigger_id,
                )
                await self._deliver(user_id, result)
            except Exception:
                log.exception("proactive: tick failed user=%s", user_id)

    async def _loop(self) -> None:
        log.info("proactive: driver started (poll=%.1fs)", self._poll_seconds)
        while self._running:
            try:
                await self._tick_once()
            except Exception:
                log.exception("proactive: tick loop iteration crashed; continuing")
            await asyncio.sleep(self._poll_seconds)
        log.info("proactive: driver stopped")

    def start(self) -> None:
        if self._running:
            return
        self._running = True
        self._task = asyncio.get_running_loop().create_task(self._loop())

    async def stop(self) -> None:
        self._running = False
        if self._task is not None:
            try:
                await asyncio.wait_for(self._task, timeout=self._poll_seconds + 2)
            except (asyncio.TimeoutError, asyncio.CancelledError):
                pass
            self._task = None

_active_users: dict[str, float] = {}  # user_id -> last-seen unix timestamp


def _touch_active_user(user_id: str) -> None:
    if user_id:
        _active_users[user_id] = time.time()


def _get_active_users() -> list[str]:
    """Users seen recently enough to be worth polling. Called once per tick."""
    cutoff = time.time() - ACTIVE_USER_TTL_S
    stale = [uid for uid, ts in _active_users.items() if ts < cutoff]
    for uid in stale:
        _active_users.pop(uid, None)
    return list(_active_users)


@app.post("/track/{user_id}")
async def track_user(user_id: str) -> dict[str, str]:
    """Fire-and-forget signal from the frontend: "this user is on the page
    right now." The proactive driver only evaluates users who've pinged
    this recently — without it there's nothing to iterate over. Call it
    once on mount/pageview, alongside `posthog.identify()` (Step 4)."""
    _touch_active_user(user_id)
    return {"status": "ok"}


async def _get_proactive_context(user_id: str) -> ProactiveTriggerContext | None:
    """Build a ProactiveTriggerContext from the REST live-activity pull —
    the on-demand equivalent of reaching into the old AsyncContextStore
    buffer. `SlimAction.from_dict` accepts the endpoint's JSON shape
    directly (title/description/canonical_url/index/type/timestamp_start/
    timestamp_end all match the wire format)."""
    raw_actions = await fetch_recent_activity_raw(user_id)
    if not raw_actions:
        return None
    slim_actions = tuple(SlimAction.from_dict(a) for a in raw_actions)
    return ProactiveTriggerContext.from_slim_actions(
        slim_actions,
        # No session concept in the REST read — ProactiveTriggerContext
        # still wants a non-empty `session_id` for its scope check, so we
        # reuse user_id there. It's just a scope string now, not a real
        # session identifier.
        session_id=user_id,
        product_id=PRODUCT_ID,
        action_count=len(slim_actions),
    )

_proactive_subscribers: dict[str, list[asyncio.Queue]] = {}


def _publish_proactive(user_id: str, event: dict) -> int:
    """Fan-out a proactive event to every live SSE subscriber for this user.
    Returns the number of subscribers that received the event."""
    queues = _proactive_subscribers.get(user_id) or []
    for q in queues:
        try:
            q.put_nowait(event)
        except asyncio.QueueFull:
            pass
    return len(queues)


# CORS — browser hits the bridge directly for the SSE stream.
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)


@app.get("/proactive/stream/{user_id}")
async def proactive_stream(
    user_id: str, request: Request,
) -> StreamingResponse:
    queue: asyncio.Queue = asyncio.Queue(maxsize=32)
    _proactive_subscribers.setdefault(user_id, []).append(queue)
    log.info("proactive: SSE subscriber connected user=%s total=%d",
             user_id, len(_proactive_subscribers[user_id]))

    async def event_gen():
        try:
            yield "retry: 3000\n\n"
            yield ": connected\n\n"
            import json as _json
            while True:
                if await request.is_disconnected():
                    break
                try:
                    event = await asyncio.wait_for(queue.get(), timeout=15.0)
                except asyncio.TimeoutError:
                    yield ": keepalive\n\n"
                    continue
                yield f"event: proactive\ndata: {_json.dumps(event)}\n\n"
        finally:
            subs = _proactive_subscribers.get(user_id) or []
            if queue in subs:
                subs.remove(queue)
            if not subs:
                _proactive_subscribers.pop(user_id, None)

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

_PROACTIVE_MESSAGES: dict[str, str] = {
    "bulk_edit_opportunity": (
        "Updating projects one by one? There's a multi-select bulk edit "
        "you might not have seen — want me to show you?"
    ),
}

# Authoritative "how-to" text per trigger. When the user clicks Open chat,
# this is fed to the LLM as ground truth so the follow-up reply has real
# UI steps instead of plausibly-hallucinated ones.
_TRIGGER_HOW_TO: dict[str, str] = {
    "bulk_edit_opportunity": (
        "How bulk edit works on the Projects page:\n"
        "1. At the top right of the table (next to the status filter), "
        "click the small kebab menu icon (the three dots: ⋯).\n"
        "2. A dropdown opens — pick 'Bulk edit'. Checkboxes appear on every "
        "row.\n"
        "3. Tick the projects you want to update.\n"
        "4. A small toolbar appears at the bottom of the screen. Pick the "
        "new priority from the dropdown.\n"
        "5. Click 'Apply to N' — all selected projects update in one shot."
    ),
}

async def _deliver_to_app(
    user_id: str, result: ProactiveTriggerResult,
) -> None:
    """Push a proactive message straight to the host app via SSE.
    Rasa is intentionally NOT in this path — proactive UX is owned by the
    app shell, not the chat vendor."""
    current_page = await _current_page_hint(user_id)  # defined in Step 8
    event = {
        "trigger_id": result.trigger_id,
        "body": _PROACTIVE_MESSAGES.get(result.trigger_id, result.body),
        "tours": _tour_payload_for_trigger(result.trigger_id, current_page),  # Step 7
    }
    delivered = _publish_proactive(user_id, event)
    log.info(
        "proactive: published user=%s trigger=%s subscribers=%d",
        user_id, result.trigger_id, delivered,
    )

_proactive: ProactiveDriver | None = None

@asynccontextmanager
async def lifespan(_: FastAPI):
    global _proactive

    _proactive = ProactiveDriver(
        registry=build_registry(),
        get_active_users=_get_active_users,
        get_context=_get_proactive_context,
        get_state=_user_state,
        deliver=_deliver_to_app,
        poll_seconds=PROACTIVE_POLL_S,
    )
    _proactive.start()

    try:
        yield
    finally:
        if _proactive is not None:
            await _proactive.stop()

"use client";

import { useEffect } from "react";
import { usePathname, useSearchParams } from "next/navigation";
import posthog from "posthog-js";

const BRIDGE_URL =
  process.env.NEXT_PUBLIC_BRIDGE_URL || "http://localhost:8090";
const USER_ID = "demo-user-1"; // same id passed to posthog.identify() below

function trackActive() {
  // Fire-and-forget — tells the bridge's ProactiveDriver this user is
  // currently active, so it's included in the next poll tick (Step 3b).
  fetch(`${BRIDGE_URL}/track/${encodeURIComponent(USER_ID)}`, {
    method: "POST",
    keepalive: true,
  }).catch(() => {});
}

export default function PostHogProvider({ children }) {
  const pathname = usePathname();
  const searchParams = useSearchParams();

  useEffect(() => {
    if (typeof window === "undefined" || posthog.__loaded) return;
    posthog.init("phc_YOUR_PROJECT_API_KEY", {
      api_host: "https://us.i.posthog.com",
      person_profiles: "identified_only",
      session_idle_timeout_seconds: 120,
      // SPA pageviews are captured manually below — PostHog's
      // `capture_pageview: true` only catches the initial load.
      capture_pageview: false,
      capture_pageleave: true,
      // For the demo: ship events immediately so the proactive pipeline
      // isn't sitting on a 30s client-side batch. Production apps keep
      // batching on.
      request_batching: false,
      loaded: (ph) => {
        ph.identify(USER_ID, {
          product_id: "YOUR_AUTOPLAY_PRODUCT_ID",
          email: "user@theirdomain.com",
        });
        ph.register({ email: "user@theirdomain.com" });
        trackActive();
      },
    });
  }, []);

  // Fire $pageview on every route change so canonical_url reflects where
  // the user actually is — not where they last clicked. Also re-touch the
  // active-user signal so a long-lived tab doesn't age out of
  // ACTIVE_USER_TTL_S between polls.
  useEffect(() => {
    if (typeof window === "undefined") return;
    const qs = searchParams?.toString();
    const url = qs ? `${pathname}?${qs}` : pathname;
    posthog.capture("$pageview", {
      $current_url: window.location.origin + url,
    });
    trackActive();
  }, [pathname, searchParams]);

  return children;
}

"use client";

import { useEffect, useState, useCallback } from "react";

const BRIDGE_URL =
  process.env.NEXT_PUBLIC_BRIDGE_URL || "http://localhost:8090";
// IMPORTANT: this MUST be the same value you pass to `posthog.identify(...)`
// in `app/posthog-provider.js` AND to `customData.userId` in
// `app/rasa-widget.js`. The bridge keys the SSE stream, the proactive FSM,
// and the chat session by this id — if they don't match, the toast never
// reaches the right tab and the LLM follow-up can't read the user's activity.
const USER_ID = "demo-user-1";
const AUTO_DISMISS_MS = 20_000;

function openChatWidget() {
  const launcher =
    document.querySelector(".rw-launcher") ||
    document.querySelector("[class*=launcher]");
  if (launcher instanceof HTMLElement) launcher.click();
}

// injectIntoChat + startTour are added in Steps 6 and 7 (top of this file).

export default function ProactiveToast() {
  const [toasts, setToasts] = useState([]);

  const dismiss = useCallback((id) => {
    setToasts((prev) => prev.filter((t) => t.id !== id));
  }, []);

  useEffect(() => {
    if (typeof window === "undefined") return;
    const source = new EventSource(
      `${BRIDGE_URL}/proactive/stream/${encodeURIComponent(USER_ID)}`
    );
    source.addEventListener("proactive", (ev) => {
      try {
        const data = JSON.parse(ev.data);
        const id = `${data.trigger_id}-${Date.now()}`;
        setToasts((prev) => [
          ...prev,
          {
            id,
            triggerId: data.trigger_id,
            body: data.body,
            tours: Array.isArray(data.tours) ? data.tours : [],
          },
        ]);
        setTimeout(() => dismiss(id), AUTO_DISMISS_MS);
      } catch {
        // malformed event — ignore
      }
    });
    return () => source.close();
  }, [dismiss]);

  if (toasts.length === 0) return null;

  return (
    <div className="pointer-events-none fixed bottom-24 right-6 z-50 flex flex-col gap-3">
      {toasts.map((t) => (
        <div
          key={t.id}
          className="pointer-events-auto w-80 rounded-xl border border-zinc-200 bg-white p-4 text-sm shadow-xl dark:border-zinc-700 dark:bg-zinc-900"
        >
          <div className="flex items-start gap-3">
            <div className="flex h-9 w-9 shrink-0 items-center justify-center rounded-full bg-indigo-600 text-white">
              💬
            </div>
            <div className="flex-1">
              <div className="text-xs font-medium uppercase tracking-wide text-zinc-500">
                Copilot
              </div>
              <div className="mt-1 leading-snug">{t.body}</div>
            </div>
            <button
              type="button"
              onClick={() => dismiss(t.id)}
              className="-mr-1 -mt-1 rounded p-1 text-zinc-400 hover:bg-zinc-100"
            >
              ×
            </button>
          </div>

          <div className="mt-3 flex flex-wrap gap-2">
            {t.tours.map((tour) => (
              <button
                key={tour.id}
                type="button"
                onClick={() => { startTour(tour); dismiss(t.id); }}
                className="rounded-md bg-indigo-600 px-3 py-1.5 text-xs font-medium text-white hover:bg-indigo-700"
              >
                {tour.label || "Show me"}
              </button>
            ))}
            <button
              type="button"
              onClick={() => {
                injectIntoChat(t.body, t.triggerId);
                dismiss(t.id);
              }}
              className="rounded-md border border-zinc-300 px-3 py-1.5 text-xs font-medium text-zinc-700 hover:bg-zinc-100"
            >
              Open chat
            </button>
          </div>
        </div>
      ))}
    </div>
  );
}

async def _push_bot_message(sender_id: str, text: str) -> None:
    """Inject a single bot utterance into the widget's open SocketIO session."""
    url = (
        f"{RASA_URL}/conversations/{sender_id}/trigger_intent"
        "?output_channel=socketio"
    )
    body = {
        "name": "EXTERNAL_proactive",
        "entities": {"proactive_text": text},
    }
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.post(url, json=body)
        r.raise_for_status()


async def _llm_reply(user_id: str, query: str) -> str:
    """Same prompt-assembly + OpenAI call /reply uses, reusable so the
    proactive follow-up can synthesise a contextual answer without a
    second HTTP hop."""
    assembly = await _build_assembly(user_id, query)
    user_prompt = build_user_prompt_block(assembly)
    name = _name_from_email(_user_emails.get(user_id))
    sys_prompt = SYSTEM_PROMPT + (
        f"\n\nUser's first name: {name}." if name else ""
    )
    resp = await _openai.chat.completions.create(
        model=LLM_MODEL,
        temperature=0.4,
        max_tokens=250,
        messages=[
            {"role": "system", "content": sys_prompt},
            {"role": "user", "content": user_prompt},
        ],
    )
    return (resp.choices[0].message.content or "").strip()

@app.post("/proactive/inject")
async def proactive_inject(payload: dict) -> dict[str, Any]:
    """User clicked "Open chat" — push the question, then auto-followup with
    an LLM answer grounded in _TRIGGER_HOW_TO.

    Frontend passes:
      - sender_id  → rasa-webchat's socket session id (from sessionStorage)
      - body       → the proactive message body (what the toast showed)
      - user_id    → the auth user id (for the LLM activity context)
      - trigger_id → which trigger fired (for the how-to lookup)
    """
    sender_id = str(payload.get("sender_id") or "")
    body = str(payload.get("body") or "")
    user_id = str(payload.get("user_id") or "")
    trigger_id = str(payload.get("trigger_id") or "")
    if not sender_id or not body:
        raise HTTPException(400, "sender_id and body required")

    # 1) Send the proactive question.
    try:
        await _push_bot_message(sender_id, body)
        log.info("proactive: injected question sender=%s len=%d",
                 sender_id, len(body))
    except Exception as exc:
        log.warning("proactive: question push failed: %s", exc)
        raise HTTPException(502, f"rasa trigger_intent failed: {exc}")

    # 2) Auto-follow-up with the LLM answer — the user clicked Open chat,
    #    that IS the "yes." Bake the proactive question AND the
    #    authoritative how-to text into the synthesised user message so
    #    the LLM has ground truth for the UI steps.
    how_to = _TRIGGER_HOW_TO.get(trigger_id, "")
    if user_id:
        try:
            synth = (
                f'Yes, please walk me through it. You just offered: '
                f'"{body}".\n\n'
            )
            if how_to:
                synth += (
                    "The exact steps for this feature in this app are:\n"
                    f"{how_to}\n\n"
                    "Rewrite those steps in a friendly tone, referencing "
                    "what I've been doing on the page. Use the activity log "
                    "to make it feel personal — but do NOT invent any "
                    "button names or UI elements that are not in the steps "
                    "above. Stay accurate to the steps."
                )
            else:
                synth += (
                    "Give me concrete, step-by-step instructions for the "
                    "feature you mentioned. Reference what I have been "
                    "doing on the page. Do NOT ask me to clarify."
                )
            followup = await _llm_reply(user_id, synth)
            if followup:
                # Small gap so the two bubbles don't render simultaneously.
                await asyncio.sleep(0.6)
                await _push_bot_message(sender_id, followup)
                log.info("proactive: injected followup sender=%s len=%d",
                         sender_id, len(followup))
        except Exception as exc:
            log.warning("proactive: followup failed: %s", exc)
            # Non-fatal — at least the question landed.

    return {"status": "ok"}

"use client";

import { useEffect } from "react";

export default function RasaWidget() {
  useEffect(() => {
    if (typeof window === "undefined") return;

    // Start every page load with a fresh chat. Run BEFORE the script-exists
    // guard so soft refreshes also clear it.
    sessionStorage.removeItem("chat_session");

    if (document.getElementById("rasa-webchat-script")) return;

    const script = document.createElement("script");
    script.id = "rasa-webchat-script";
    script.src =
      "https://cdn.jsdelivr.net/npm/rasa-webchat@1.0.1/lib/index.js";
    script.async = true;
    script.onload = () => {
      window.WebChat.default(
        {
          customData: { language: "en", userId: "demo-user-1" },
          socketUrl: "http://localhost:5005",
          title: "Demo Copilot",
          subtitle: "Powered by Autoplay + Rasa",
          inputTextFieldHint: "Ask me what you just did, or for help...",
          showFullScreenButton: false,
          params: { storage: "session" },
        },
        null,
      );
    };
    document.body.appendChild(script);
  }, []);

  return null;
}

def _mark_reactive(user_id: str) -> None:
    """Tell the FSM the user just sent a chat message.

    Two SDK calls per the agent-state-v2 contract:
      - `transition_to_reactive()` moves THINKING → REACTIVE so future
        proactive triggers are gated until the conversation ends.
      - `record_user_interaction()` resets `interaction_timeout_s` so
        the FSM doesn't drift back to THINKING mid-conversation.
    """
    state = _user_state(user_id)
    try:
        if state.current_state == AgentStateV2.THINKING:
            state.transition_to_reactive()
    except Exception as exc:
        log.debug("FSM transition_to_reactive skipped: %s", exc)
    try:
        state.record_user_interaction()
    except Exception as exc:
        log.debug("FSM record_user_interaction skipped: %s", exc)


@app.get("/reply/{user_id}")
async def get_reply(user_id: str, query: str, email: str | None = None) -> dict[str, Any]:
    if not query:
        raise HTTPException(status_code=400, detail="query required")
    _mark_reactive(user_id)   # ← new
    # ... rest of the Step 1 implementation ...

"use client";

import { useEffect } from "react";
import usertour from "usertour.js";

const USERTOUR_TOKEN = "YOUR_USERTOUR_TOKEN";
const USER_ID = "demo-user-1";
const BRIDGE_URL =
  process.env.NEXT_PUBLIC_BRIDGE_URL || "http://localhost:8090";

function postTourEvent(path) {
  fetch(`${BRIDGE_URL}/tour/${path}`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ user_id: USER_ID }),
    keepalive: true,
  }).catch(() => {
    // non-fatal — the FSM will time out on its own if we miss the call
  });
}

// Usertour's event API is not strongly typed in the npm wrapper; try a few
// common shapes and quietly skip what's not there.
function wireTourEventCallbacks(u) {
  const tryOn = (event, handler) => {
    if (typeof u.on === "function") {
      try { u.on(event, handler); } catch { /* ignore */ }
    }
  };
  tryOn("flow:step:complete", () => postTourEvent("step"));
  tryOn("flow:step:advance",  () => postTourEvent("step"));
  tryOn("step:complete",      () => postTourEvent("step"));
  tryOn("flow:complete",      () => postTourEvent("end"));
  tryOn("flow:end",           () => postTourEvent("end"));
  tryOn("flow:dismiss",       () => postTourEvent("end"));
}

export default function UsertourProvider() {
  useEffect(() => {
    if (typeof window === "undefined") return;
    try {
      usertour.init(USERTOUR_TOKEN);
    } catch (e) {
      console.warn("[usertour] init failed (non-fatal):", e);
      return;
    }
    Promise.resolve(
      usertour.identify(USER_ID, {
        email: "user@theirdomain.com",
        name: "Demo User",
      })
    ).catch((e) =>
      console.warn("[usertour] identify failed (non-fatal):", e?.message || e)
    );
    if (window.usertour) wireTourEventCallbacks(window.usertour);
  }, []);
  return null;
}

TOUR_REGISTRY = TourRegistry.from_dict({
    "product_id": "YOUR_AUTOPLAY_PRODUCT_ID",
    "default_interaction_timeout_s": 60.0,
    "default_cooldown_period_s": 120.0,
    "tours": [
        {
            "id": "projects-bulk-edit",
            "user_tour_id": "YOUR_USERTOUR_FLOW_ID",  # from Step 7a step 7
            "user_tour_name": "Projects bulk edit",
        },
    ],
})

# Which tours are offered as a "Show me" CTA when each trigger fires.
_TRIGGER_TOURS: dict[str, list[str]] = {
    "bulk_edit_opportunity": ["projects-bulk-edit"],
}

# A tour only runs on pages where its anchors exist. Substring match against
# the user's current canonical URL.
_TOUR_PAGE_PATTERNS: dict[str, str] = {
    "projects-bulk-edit": "/projects",
}


def _tour_payload_for_trigger(
    trigger_id: str, current_page: str | None,
) -> list[dict]:
    """Return [{id, flow_id, label}] for tours to surface on this trigger.
    Tours are filtered by `current_page` — a tour's anchors only exist on
    its target page, so offering "Show me" anywhere else spotlights nothing."""
    if not current_page:
        return []
    out: list[dict] = []
    for tour_id in _TRIGGER_TOURS.get(trigger_id, []):
        page_pattern = _TOUR_PAGE_PATTERNS.get(tour_id)
        if page_pattern and page_pattern not in current_page:
            continue
        tour_def = TOUR_REGISTRY.get(tour_id)
        if tour_def is None or not tour_def.user_tour_id:
            continue
        out.append({
            "id": tour_def.id,
            "flow_id": tour_def.user_tour_id,
            "label": "Show me",
        })
    return out

@app.post("/tour/start")
async def tour_start(payload: dict) -> dict[str, Any]:
    """User accepted the "Show me" CTA; return the Usertour flow_id and
    transition the FSM into visual-guidance mode."""
    user_id = str(payload.get("user_id") or "")
    tour_id = str(payload.get("tour_id") or "")
    if not user_id or not tour_id:
        raise HTTPException(400, "user_id and tour_id required")
    tour_def = TOUR_REGISTRY.get(tour_id)
    if tour_def is None or not tour_def.user_tour_id:
        raise HTTPException(404, f"unknown tour {tour_id}")

    state = _user_state(user_id)
    try:
        state.set_visual_guidance(active=True, tour_id=tour_id)
        log.info("tour: started user=%s tour=%s flow=%s",
                 user_id, tour_id, tour_def.user_tour_id)
    except Exception as exc:
        log.warning("tour: set_visual_guidance failed: %s", exc)

    return {"status": "ok", "flow_id": tour_def.user_tour_id, "tour_id": tour_id}


@app.post("/tour/step")
async def tour_step(payload: dict) -> dict[str, str]:
    """Frontend pings here on each tour step so the FSM's interaction timer
    keeps resetting and proactive triggers stay gated."""
    user_id = str(payload.get("user_id") or "")
    if not user_id:
        return {"status": "no_user"}
    state = _user_state(user_id)
    try:
        state.record_tour_step()
    except Exception as exc:
        log.warning("tour: record_tour_step failed: %s", exc)
    return {"status": "ok"}


@app.post("/tour/end")
async def tour_end(payload: dict) -> dict[str, str]:
    """Tour completed or dismissed — clear the visual-guidance flag."""
    user_id = str(payload.get("user_id") or "")
    if not user_id:
        return {"status": "no_user"}
    state = _user_state(user_id)
    try:
        state.set_visual_guidance(active=False)
        log.info("tour: ended user=%s", user_id)
    except Exception as exc:
        log.warning("tour: set_visual_guidance(end) failed: %s", exc)
    return {"status": "ok"}

// Best-effort: try the public methods Usertour exposes for programmatic start.
// `usertour.js` has shipped multiple shapes — pin the version you tested OR
// keep this fallthrough so customers on older / newer minor versions still work.
function startUsertourFlow(flowId) {
  const u = window.usertour;
  if (!u) return false;
  if (typeof u.start === "function") {
    u.start(flowId);
    return true;
  }
  if (u.flow && typeof u.flow.start === "function") {
    u.flow.start(flowId);
    return true;
  }
  if (u.flows && typeof u.flows.start === "function") {
    u.flows.start(flowId);
    return true;
  }
  return false;
}

async function startTour(tour) {
  // Flip the FSM into visual-guidance mode + log on the bridge.
  try {
    await fetch(`${BRIDGE_URL}/tour/start`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ user_id: USER_ID, tour_id: tour.id }),
    });
  } catch {
    // non-fatal — still run the tour
  }
  const launched = startUsertourFlow(tour.flow_id);
  if (!launched) {
    console.warn(
      "[proactive] Usertour SDK not ready; could not start flow",
      tour.flow_id,
    );
  }
}

async def _current_page_hint(user_id: str) -> str | None:
    """Latest URL the user is on — picked by `timestamp_start`, not array
    order. Reuses the same REST pull `_get_proactive_context` uses; if
    you're calling both for the same tick, consider fetching once and
    passing the raw actions to both instead of two REST round-trips.
    """
    actions = await fetch_recent_activity_raw(user_id)
    latest_url = None
    latest_ts = float("-inf")
    for action in actions:
        url = action.get("canonical_url")
        if not url:
            continue
        try:
            ts = float(action.get("timestamp_start") or 0.0)
        except (TypeError, ValueError):
            ts = 0.0
        if ts > latest_ts:
            latest_ts = ts
            latest_url = url
    return latest_url