From Reactive Chatbots to Always-On AI: Building Event-Driven Shopping Assistants

┌─────────────────┐ WebSocket  ┌──────────────────┐
│    Browser      │───────────>│  Event Gateway   │
│  (user events)  │            │   (ingestion)    │
└────────┬────────┘            └────────┬─────────┘
         │                              │
         │ WebSocket (actions)          ▼
         │◄────────────────────  ┌──────────────┐
         │                       │  Event Bus   │
         │                       │ (Kafka/NATS) │
         │                       └──────┬───────┘
         │                              │
         │                              ▼
         │                       ┌─────────────────┐
         │                       │  Agent Runtime  │
         │◄──────────────────────│ (LLM orchestr.) │
         │   Agent Commands      └─────────────────┘

{
  "source": "browser",
  "type": "LongDwellOnDeliveryOption",
  "userId": "u482",
  "sessionId": "sess_123",
  "payload": {
    "currentStep": "delivery-selection",
    "dwellTime": 47000,
    "mouseOscillation": "high"
  },
  "timestamp": "2025-01-13T20:51:03Z"
}

{
  "source": "inventory",
  "type": "stock.below_threshold",
  "productId": "prod_789",
  "variant": "size_M",
  "stock": 3,
  "threshold": 5,
  "timestamp": "2025-01-13T20:51:00Z"
}

{
  "source": "stripe",
  "type": "payment.provider_degraded",
  "provider": "card_payments",
  "degradationLevel": "moderate",
  "alternativeProvider": "paypal",
  "timestamp": "2025-01-13T20:50:45Z"
}

interface BrowserEvent {
  type: string;           // 'page.view' | 'form.input' | 'basket.update'
  timestamp: string;      // ISO 8601
  sessionId: string;
  userId?: string;
  payload: Record<string, any>;
}

{
  "type": "ui.suggestion",
  "text": "Delivery slots around 6–8pm are usually cheapest. Want me to reserve one?",
  "priority": "medium",
  "actionable": true
}

{
  "type": "dom.fill",
  "selector": "#address-line-1",
  "value": "123 Main Street",
  "annotation": "Using your saved home address"
}

{
  "type": "prefetch.deliverySlots",
  "payload": { "postcode": "SW1A 1AA" }
}

{
  "type": "workflow.start",
  "name": "delivery_estimator",
  "steps": [
    "open_delivery_menu",
    "fetch_slots",
    "highlight_best_slot"
  ]
}

{"event": "slot_capacity_changed", "zone": "Z3", "from": 9, "to": 10}
{"event": "slot_indexed", "zone": "Z3", "slot": "18:00"}
{"event": "slot_reconciliation", "zone": "Z3"}

{
  "type": "delivery.slot_opened",
  "slot": "18:00-20:00",
  "zone": "Z3",
  "change": "added",
  "priority": "medium",
  "reason": "new_capacity"
}

┌────────────────┐
│  Raw Events    │  (thousands/sec, low meaning)
└────────┬───────┘
         │
         ▼
┌──────────────────────────┐
│ Signal Derivation Layer  │
│  - Aggregation           │
│  - Deduplication         │
│  - Threshold filtering   │
│  - Temporal stitching    │
└────────┬─────────────────┘
         │
         ▼ (few/sec, high meaning)
┌────────────────────┐
│ Relevance Router   │
│  - User matching   │
│  - Priority scoring│
│  - Cohort filtering│
└────────┬───────────┘
         │
         ▼ (1-6 users)
┌────────────────┐
│  Agent Runtime │
└────────────────┘

slot_updated zone=Z3 capacity=9→10
slot_updated zone=Z3 capacity=10→10
slot_indexed zone=Z3 slot=18:00
slot_indexed zone=Z3 slot=18:00
...

{
  "type": "delivery.slot_opened",
  "slot": "18:00-20:00",
  "zone": "Z3",
  "impactedPostcodes": ["11000", "11001", "11002"],
  "estimatedDemand": "high",
  "priority": "medium"
}

function calculateRelevanceScore(user: User, signal: DeliverySlotSignal): number {
  let score = 0;

  // Geographic match (required)
  if (!user.postcode.matchesZone(signal.zone)) return 0;

  // Currently viewing delivery options (high value)
  if (user.currentStep === 'delivery-selection') score += 50;

  // Dwelling/stuck (indicates need)
  if (user.dwellTime > 30000) score += 30;

  // Historical preference match
  if (user.preferredSlotTime === signal.slot.startTime) score += 20;

  // Previously had no slots available
  if (user.sessionEvents.includes('no_slots_available')) score += 40;

  // Business rules
  if (user.tier === 'vip') score += 25;
  if (user.cartValue > 100) score += 15;

  return score;
}

{
  "type": "delivery.slot_offered",
  "slot": "18:00-20:00",
  "zone": "Z3",
  "userId": "u482",
  "reservationExpiry": "2025-01-13T20:53:00Z" // 90 second soft hold
}

// Find all users impacted by a slot change
MATCH (zone:Zone {id: 'Z3'})-[:SERVES]->(postcode:Postcode)
MATCH (user:User)-[:LIVES_IN]->(postcode)
MATCH (user)-[:IN_SESSION]->(:Session {active: true})
WHERE user.currentStep = 'delivery-selection'
RETURN user.id, user.tier, user.cartValue
ORDER BY user.tier DESC, user.cartValue DESC

MATCH (product:Product {id: 'prod_789'})-[:VARIANT]->(variant:Variant {size: 'M'})
MATCH (user:User)-[:VIEWED]->(variant)
WHERE user.lastView > datetime() - duration('PT10M')
RETURN user.id, user.sessionId

{
  "type": "inventory.scarcity_alert",
  "product": "prod_789",
  "variant": "size_M",
  "stock": 3,
  "urgency": "high",
  "reason": "popular_item_low_stock"
}

interface AgentContext {
  user: {
    id: string;
    tier: 'standard' | 'vip';
    sessionDuration: number;
    cartValue: number;
    currentStep: string;
  };

  recentEvents: Array<{
    type: string;
    timestamp: string;
    payload: any;
  }>;

  signals: Array<{
    type: string;
    priority: 'low' | 'medium' | 'high';
    source: string;
    payload: any;
  }>;

  sessionHistory: {
    pageViews: number;
    productsViewed: number;
    cartAdditions: number;
    checkoutAttempts: number;
    previousPurchases: number;
  };

  constraints: {
    notificationsSentToday: number;
    lastInteractionTime: string;
    userPreferences: {
      proactiveAssistance: boolean;
      notificationFrequency: 'minimal' | 'normal' | 'maximum';
    };
  };
}

- Is it relevant to user's current context?
- Is the timing appropriate?
- Have I already suggested something similar?

- Proactive: Show suggestion without user asking
- Reactive: Prepare answer but wait for user to initiate

- High urgency: "Only 2 left in your size - reserve now?"
- Medium: "A cheaper delivery slot just opened for tonight"
- Low: "By the way, you forgot eggs last time"

- Max 3 proactive suggestions per session
- Min 5 minutes between unprompted messages
- Respect user's notification preferences

Confidence: HIGH (4 positive signals)
Urgency: MEDIUM-HIGH (stock scarcity + user interest)
Action: Proactive suggestion
Tone: Helpful, time-sensitive
Throttle: Allow (0 suggestions sent this session)

Response: "Size M is running low (only 3 left). Want me to reserve
one for the next 10 minutes while you finish browsing?"

You are a shopping assistant for an e-commerce platform. Your goal is to help users complete purchases without being annoying.

CONTEXT:
- User is on step: {currentStep}
- Cart value: ${cartValue}
- Session duration: {sessionDuration} minutes
- Notifications sent today: {notificationCount}

RECENT EVENTS:
{recentEvents}

NEW SIGNALS:
{signals}

POLICY:
- Maximum 3 proactive suggestions per session
- Minimum 5 minutes between suggestions
- Only suggest when confidence > 70%
- Respect user preference: {notificationFrequency}

TASK:
Evaluate the signals and decide:
1. Should you send a proactive message? (yes/no)
2. If yes, what should it say?
3. What action should it enable?
4. What priority level? (low/medium/high)

Output as JSON:
{
  "shouldAct": boolean,
  "message": string,
  "action": {
    "type": string,
    "payload": object
  },
  "priority": string,
  "reasoning": string
}

{
  "type": "ui.suggestion",
  "priority": "high",
  "message": "This organic milk is popular and only 2 left. Want me to add it to your cart?",
  "action": {
    "type": "cart.add_item",
    "productId": "prod_789",
    "quantity": 1
  }
}

{
  "type": "ui.suggestion",
  "priority": "medium",
  "message": "A 6-8pm slot just opened (usually $3 cheaper than morning). Should I grab it for you?",
  "action": {
    "type": "delivery.reserve_slot",
    "slotId": "slot_18_20",
    "duration": 90
  }
}

{
  "type": "ui.notification",
  "priority": "high",
  "message": "Our card payment system is experiencing delays. PayPal is working perfectly - want to switch?",
  "action": {
    "type": "payment.switch_provider",
    "toProvider": "paypal"
  }
}

{
  "type": "ui.suggestion",
  "priority": "low",
  "message": "I noticed you usually order eggs - want to add them? Your favorite brand is in stock.",
  "action": {
    "type": "cart.suggest_item",
    "productId": "prod_eggs_organic",
    "reason": "frequently_purchased"
  }
}

{
  "type": "ui.suggestion",
  "priority": "medium",
  "message": "I noticed you've been browsing dairy-free items. Want me to filter out all dairy products from your results?",
  "action": {
    "type": "preferences.set_dietary_filter",
    "filter": "dairy_free",
    "persistent": true
  }
}

// event-capture-worker.ts
const eventQueue: BrowserEvent[] = [];
const BATCH_SIZE = 10;
const BATCH_INTERVAL = 1000; // 1 second

self.addEventListener('message', (e) => {
  const event = e.data;
  eventQueue.push(event);

  if (eventQueue.length >= BATCH_SIZE) {
    flushEvents();
  }
});

setInterval(flushEvents, BATCH_INTERVAL);

function flushEvents() {
  if (eventQueue.length === 0) return;

  const batch = eventQueue.splice(0, BATCH_SIZE);

  fetch('https://events.example.com/batch', {
    method: 'POST',
    body: JSON.stringify(batch)
  });
}

// Main thread
const worker = new Worker('event-capture-worker.js');

// Page view
document.addEventListener('DOMContentLoaded', () => {
  worker.postMessage({
    type: 'page.view',
    url: window.location.href,
    referrer: document.referrer,
    timestamp: new Date().toISOString()
  });
});

// Scroll depth
let maxScroll = 0;
window.addEventListener('scroll', debounce(() => {
  const scrollPercent = (window.scrollY / document.body.scrollHeight) * 100;
  if (scrollPercent > maxScroll) {
    maxScroll = scrollPercent;
    worker.postMessage({
      type: 'scroll.depth',
      percent: Math.round(scrollPercent),
      timestamp: new Date().toISOString()
    });
  }
}, 500));

// Dwell detection
let dwellTimer: number;
const dwellThreshold = 5000; // 5 seconds

document.addEventListener('mouseenter', () => {
  dwellTimer = setTimeout(() => {
    worker.postMessage({
      type: 'user.dwelling',
      element: document.activeElement?.id,
      duration: dwellThreshold,
      timestamp: new Date().toISOString()
    });
  }, dwellThreshold);
});

document.addEventListener('mouseleave', () => {
  clearTimeout(dwellTimer);
});

// Cart state changes
window.addEventListener('cart:updated', (e: CustomEvent) => {
  worker.postMessage({
    type: 'cart.updated',
    itemCount: e.detail.itemCount,
    cartValue: e.detail.total,
    timestamp: new Date().toISOString()
  });
});

-- Create event log
CREATE TABLE event_log (
  id SERIAL PRIMARY KEY,
  type TEXT NOT NULL,
  source TEXT NOT NULL,
  payload JSONB NOT NULL,
  user_id TEXT,
  session_id TEXT,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Notify on insert
CREATE OR REPLACE FUNCTION notify_event() RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify('events', row_to_json(NEW)::text);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER event_notify
AFTER INSERT ON event_log
FOR EACH ROW EXECUTE FUNCTION notify_event();

import { Redis } from 'ioredis';

const redis = new Redis();

// Publisher
async function publishEvent(event: Event) {
  await redis.xadd(
    'events',
    '*',
    'type', event.type,
    'source', event.source,
    'payload', JSON.stringify(event.payload),
    'userId', event.userId || '',
    'timestamp', event.timestamp
  );
}

// Consumer
async function consumeEvents(callback: (event: Event) => void) {
  const streams = ['events'];
  let lastId = '$'; // Start from now

  while (true) {
    const results = await redis.xread(
      'BLOCK', 0,
      'STREAMS', ...streams, lastId
    );

    if (results) {
      for (const [stream, messages] of results) {
        for (const [id, fields] of messages) {
          const event = parseEvent(fields);
          callback(event);
          lastId = id;
        }
      }
    }
  }
}

import { Kafka } from 'kafkajs';

const kafka = new Kafka({
  clientId: 'event-system',
  brokers: ['localhost:9092']
});

// Publisher
const producer = kafka.producer();

async function publishEvent(event: Event) {
  await producer.send({
    topic: 'user-events',
    messages: [{
      key: event.userId,
      value: JSON.stringify(event),
      headers: {
        'event-type': event.type,
        'event-source': event.source
      }
    }]
  });
}

// Consumer
const consumer = kafka.consumer({ groupId: 'agent-runtime' });

await consumer.subscribe({ topic: 'user-events' });

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value.toString());
    await processEvent(event);
  }
});

// signal-deriver.ts
import { EventEmitter } from 'events';

interface RawEvent {
  type: string;
  source: string;
  payload: any;
  timestamp: string;
}

interface Signal {
  type: string;
  priority: 'low' | 'medium' | 'high';
  payload: any;
  derivedFrom: string[];
}

class SignalDeriver extends EventEmitter {
  private eventBuffer: Map<string, RawEvent[]> = new Map();
  private dedupeWindow = 3000; // 3 seconds

  processRawEvent(event: RawEvent) {
    const key = `${event.type}:${JSON.stringify(event.payload)}`;

    if (!this.eventBuffer.has(key)) {
      this.eventBuffer.set(key, []);

      // Set timeout to process buffer
      setTimeout(() => {
        this.deriveSignal(key);
      }, this.dedupeWindow);
    }

    this.eventBuffer.get(key)!.push(event);
  }

  private deriveSignal(bufferKey: string) {
    const events = this.eventBuffer.get(bufferKey) || [];
    this.eventBuffer.delete(bufferKey);

    if (events.length === 0) return;

    // Example: Delivery slot events
    if (events[0].type === 'delivery.slot_capacity_changed') {
      const signal: Signal = {
        type: 'delivery.slot_opened',
        priority: 'medium',
        payload: {
          zone: events[0].payload.zone,
          slot: events[0].payload.slot,
          changeCount: events.length,
          finalCapacity: events[events.length - 1].payload.capacity
        },
        derivedFrom: events.map(e => e.type)
      };

      this.emit('signal', signal);
    }

    // Example: Stock scarcity
    if (events[0].type === 'inventory.stock_update') {
      const latestStock = events[events.length - 1].payload.stock;

      if (latestStock <= 5) {
        const signal: Signal = {
          type: 'inventory.scarcity_alert',
          priority: latestStock <= 2 ? 'high' : 'medium',
          payload: {
            productId: events[0].payload.productId,
            variant: events[0].payload.variant,
            stock: latestStock,
            trend: this.calculateStockTrend(events)
          },
          derivedFrom: events.map(e => e.type)
        };

        this.emit('signal', signal);
      }
    }
  }

  private calculateStockTrend(events: RawEvent[]): 'declining' | 'stable' | 'increasing' {
    if (events.length < 2) return 'stable';

    const first = events[0].payload.stock;
    const last = events[events.length - 1].payload.stock;

    if (last < first) return 'declining';
    if (last > first) return 'increasing';
    return 'stable';
  }
}

// Usage
const deriver = new SignalDeriver();

deriver.on('signal', (signal: Signal) => {
  console.log('Derived signal:', signal);
  // Pass to relevance router
  routeSignalToRelevantUsers(signal);
});

// Feed raw events
consumeEvents((event) => {
  deriver.processRawEvent(event);
});

// agent-runtime.ts
import Anthropic from '@anthropic-ai/sdk';

const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY
});

interface AgentDecision {
  shouldAct: boolean;
  message?: string;
  action?: {
    type: string;
    payload: any;
  };
  priority?: 'low' | 'medium' | 'high';
  reasoning: string;
}

async function evaluateSignal(
  signal: Signal,
  user: User,
  context: AgentContext
): Promise<AgentDecision> {
  const prompt = buildAgentPrompt(signal, user, context);

  const message = await anthropic.messages.create({
    model: 'claude-3-5-sonnet-20241022',
    max_tokens: 1024,
    messages: [{
      role: 'user',
      content: prompt
    }]
  });

  const decision: AgentDecision = JSON.parse(
    message.content[0].type === 'text' ? message.content[0].text : '{}'
  );

  // Validation and safety checks
  if (decision.shouldAct) {
    // Check frequency limits
    if (context.constraints.notificationsSentToday >= 3) {
      decision.shouldAct = false;
      decision.reasoning = 'Frequency limit exceeded';
    }

    // Check time since last interaction
    const timeSinceLastInteraction =
      Date.now() - new Date(context.constraints.lastInteractionTime).getTime();

    if (timeSinceLastInteraction < 5 * 60 * 1000) { // 5 minutes
      decision.shouldAct = false;
      decision.reasoning = 'Too soon since last interaction';
    }
  }

  return decision;
}

function buildAgentPrompt(
  signal: Signal,
  user: User,
  context: AgentContext
): string {
  return `You are a shopping assistant. Evaluate this signal and decide how to help.

USER CONTEXT:
- Current step: ${user.currentStep}
- Cart value: $${context.user.cartValue}
- Session duration: ${context.user.sessionDuration} minutes
- Tier: ${user.tier}
- Notifications today: ${context.constraints.notificationsSentToday}/3

SIGNAL:
Type: ${signal.type}
Priority: ${signal.priority}
Payload: ${JSON.stringify(signal.payload, null, 2)}

RECENT EVENTS (last 5):
${context.recentEvents.slice(-5).map(e => `- ${e.type} at ${e.timestamp}`).join('\n')}

SESSION HISTORY:
- Products viewed: ${context.sessionHistory.productsViewed}
- Cart additions: ${context.sessionHistory.cartAdditions}
- Previous purchases: ${context.sessionHistory.previousPurchases}

POLICY:
- Max 3 proactive messages per session
- Min 5 minutes between messages
- Respect user preference: ${context.constraints.userPreferences.notificationFrequency}
- Only act if confidence > 70%

TASK:
Decide whether to send a proactive message and what it should say.

Respond in JSON:
{
  "shouldAct": boolean,
  "message": "string (if shouldAct is true)",
  "action": {
    "type": "string",
    "payload": {}
  },
  "priority": "low|medium|high",
  "reasoning": "explain your decision"
}`;
}

// Execute the decision
async function executeAgentDecision(
  decision: AgentDecision,
  userId: string
) {
  if (!decision.shouldAct) {
    console.log(`[${userId}] No action taken:`, decision.reasoning);
    return;
  }

  // Send command to browser via WebSocket
  const ws = getWebSocketForUser(userId);

  ws.send(JSON.stringify({
    type: decision.action?.type,
    message: decision.message,
    priority: decision.priority,
    payload: decision.action?.payload
  }));

  // Log interaction
  await logInteraction({
    userId,
    type: 'proactive_message',
    message: decision.message,
    action: decision.action,
    timestamp: new Date().toISOString()
  });
}

// Instead of WebSocket for inbound channel
app.get('/api/actions/stream', (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  const userId = req.query.userId;

  // Subscribe to user-specific actions
  subscribeToUserActions(userId, (action) => {
    res.write(`data: ${JSON.stringify(action)}\n\n`);
  });
});

const eventBatcher = {
  queue: [] as Event[],
  timeout: null as NodeJS.Timeout | null,

  add(event: Event) {
    this.queue.push(event);

    if (!this.timeout) {
      this.timeout = setTimeout(() => {
        this.flush();
      }, 100); // 100ms batch window
    }

    // Force flush if queue is large
    if (this.queue.length >= 20) {
      this.flush();
    }
  },

  flush() {
    if (this.queue.length === 0) return;

    sendEventBatch(this.queue);
    this.queue = [];

    if (this.timeout) {
      clearTimeout(this.timeout);
      this.timeout = null;
    }
  }
};

// DON'T capture everything
const badEvent = {
  type: 'form.input',
  field: 'credit-card',
  value: '4532-1234-5678-9010', // ❌ Never capture PII
  timestamp: '2025-01-13T20:51:00Z'
};

// DO capture intent without sensitive data
const goodEvent = {
  type: 'checkout.step_completed',
  step: 'payment',
  method: 'card', // ✓ High-level intent only
  timestamp: '2025-01-13T20:51:00Z'
};

// Use WSS (WebSocket Secure)
const ws = new WebSocket('wss://events.example.com/client', {
  headers: {
    'Authorization': `Bearer ${authToken}`
  }
});

// Validate tokens server-side
wss.on('connection', (ws, req) => {
  const token = req.headers['authorization']?.split(' ')[1];

  const user = verifyToken(token);
  if (!user) {
    ws.close(1008, 'Unauthorized');
    return;
  }

  // Associate connection with user
  connections.set(user.id, ws);
});

-- Auto-expire events after 30 days
CREATE TABLE event_log (
  id SERIAL PRIMARY KEY,
  type TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Automated cleanup
CREATE INDEX idx_event_log_created_at ON event_log(created_at);

-- Run daily
DELETE FROM event_log WHERE created_at < NOW() - INTERVAL '30 days';