sab kam khatam guys heehe

.gitignore

@@ -1 +1,3 @@
 __pycache__/
+
+.env

agent/extraction_agent.py

@@ -1,70 +1,69 @@
 """
-Field extraction agent for Notiflow.
-"""
+extraction_agent.py
+-------------------
+Stage 3: Field Extraction Agent for Notiflow
+
+Uses ModelRouter (Nova primary → Gemini fallback) to extract structured
+business fields from Hinglish messages, given a pre-classified intent.
 
-from __future__ import annotations
+    message → Intent Agent → intent → Extraction Agent → structured fields
+
+Integration note (backend upgrade):
+    The private _call_model() function now delegates to agent/model_router.py.
+    All schema enforcement, parsing, and public API logic is unchanged.
+"""
 
 import json
 import logging
 import re
 from pathlib import Path
 
-from botocore.exceptions import BotoCoreError, ClientError
-
-from app.bedrock_client import get_bedrock_client
-from app.config import MODEL_ID
-
 PROMPT_PATH = Path(__file__).parent.parent / "prompts" / "extraction_prompt.txt"
 
 INTENT_SCHEMA: dict[str, list[str]] = {
-    "order": ["intent", "customer", "item", "quantity"],
-    "payment": ["intent", "customer", "amount", "payment_type"],
-    "credit": ["intent", "customer", "item", "quantity", "amount"],
-    "return": ["intent", "customer", "item", "reason"],
+    "order":       ["intent", "customer", "item", "quantity"],
+    "payment":     ["intent", "customer", "amount", "payment_type"],
+    "credit":      ["intent", "customer", "item", "quantity", "amount"],
+    "return":      ["intent", "customer", "item", "reason"],
     "preparation": ["intent", "item", "quantity"],
-    "other": ["intent", "note"],
+    "other":       ["intent", "note"],
 }
-
 VALID_INTENTS = set(INTENT_SCHEMA.keys())
-logger = logging.getLogger(__name__)
 
+logger = logging.getLogger(__name__)
 
-def _get_bedrock_client():
-    """Return a cached Bedrock runtime client."""
-    return get_bedrock_client()
 
+# ---------------------------------------------------------------------------
+# Prompt loader
+# ---------------------------------------------------------------------------
 
 def _load_prompt(message: str, intent: str) -> str:
-    """Load the extraction prompt template and inject the message and intent."""
     template = PROMPT_PATH.read_text(encoding="utf-8")
-    prompt = template.replace("{message}", message.strip())
-    return prompt.replace("{intent}", intent.strip().lower())
-
-
-def _call_nova(prompt: str) -> str:
-    """Send a prompt to Amazon Nova 2 Lite via Bedrock Converse API."""
-    client = _get_bedrock_client()
-    request_body = {
-        "messages": [{"role": "user", "content": [{"text": prompt}]}],
-        "inferenceConfig": {
-            "maxTokens": 256,
-            "temperature": 0.0,
-            "topP": 1.0,
-        },
-    }
+    prompt   = template.replace("{message}", message.strip())
+    prompt   = prompt.replace("{intent}",  intent.strip().lower())
+    return prompt
 
-    try:
-        response = client.converse(modelId=MODEL_ID, **request_body)
-        output_message = response["output"]["message"]
-        text_parts = [block["text"] for block in output_message["content"] if "text" in block]
-        return " ".join(text_parts).strip()
-    except (BotoCoreError, ClientError) as exc:
-        logger.error("Bedrock API error: %s", exc)
-        raise RuntimeError(f"Failed to call Amazon Nova: {exc}") from exc
 
+# ---------------------------------------------------------------------------
+# Model inference  (now via ModelRouter — Nova primary, Gemini fallback)
+# ---------------------------------------------------------------------------
+
+def _call_model(prompt: str) -> str:
+    """
+    Route the prompt through ModelRouter.
+    Returns raw text response from whichever model was available.
+    """
+    from agent.model_router import route
+    raw, model_used = route(prompt, max_tokens=256)
+    logger.info("Extraction inference served by: %s", model_used)
+    return raw
+
+
+# ---------------------------------------------------------------------------
+# Response parser & normaliser
+# ---------------------------------------------------------------------------
 
 def _parse_extraction_response(raw: str, intent: str) -> dict:
-    """Parse model output into a schema-conformant dict."""
     cleaned = re.sub(r"```(?:json)?|```", "", raw).strip()
 
     try:
@@ -75,62 +74,70 @@ def _parse_extraction_response(raw: str, intent: str) -> dict:
             try:
                 parsed = json.loads(match.group(0))
             except json.JSONDecodeError:
-                logger.warning("Could not parse Nova response as JSON; returning nulls")
+                logger.warning("Could not parse model response as JSON; returning nulls")
                 parsed = {}
         else:
             parsed = {}
 
     schema_fields = INTENT_SCHEMA.get(intent, INTENT_SCHEMA["other"])
-    result = {field: parsed.get(field, None) for field in schema_fields}
+    result        = {field: parsed.get(field, None) for field in schema_fields}
     result["intent"] = intent
 
     if "customer" in result and isinstance(result["customer"], str):
         result["customer"] = result["customer"].strip().title()
 
-    if "amount" in result and result["amount"] is not None:
-        try:
-            result["amount"] = float(result["amount"])
-            if result["amount"].is_integer():
-                result["amount"] = int(result["amount"])
-        except (ValueError, TypeError):
-            result["amount"] = None
-
-    if "quantity" in result and result["quantity"] is not None:
-        try:
-            result["quantity"] = float(result["quantity"])
-            if result["quantity"].is_integer():
-                result["quantity"] = int(result["quantity"])
-        except (ValueError, TypeError):
-            result["quantity"] = None
+    for num_field in ("amount", "quantity"):
+        if num_field in result and result[num_field] is not None:
+            try:
+                val = float(result[num_field])
+                result[num_field] = int(val) if val.is_integer() else val
+            except (ValueError, TypeError):
+                result[num_field] = None
 
     return result
 
 
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
 def extract_fields(message: str, intent: str) -> dict:
-    """Extract structured business fields from a Hinglish message."""
+    """
+    Extract structured business fields from a Hinglish message.
+
+    Args:
+        message: Raw business message (Hinglish or English).
+        intent:  Intent string from the Intent Agent.
+
+    Returns:
+        Dict with "intent" + extracted fields. Missing fields are null.
+
+    Raises:
+        ValueError:   Unsupported intent.
+        RuntimeError: Both Nova and Gemini unavailable.
+
+    Examples:
+        >>> extract_fields("rahul ne 15000 bheja", "payment")
+        {'intent': 'payment', 'customer': 'Rahul', 'amount': 15000, 'payment_type': None}
+    """
     if not message or not message.strip():
-        logger.warning("Empty message received")
         return _null_result(intent)
 
     intent = intent.lower().strip()
     if intent not in VALID_INTENTS:
-        raise ValueError(
-            f"Unsupported intent: '{intent}'. Must be one of: {', '.join(sorted(VALID_INTENTS))}"
-        )
+        raise ValueError(f"Unsupported intent: '{intent}'.")
 
-    logger.info("Extracting fields | intent=%s | message=%r", intent, message)
+    logger.info("Extracting | intent=%s | message=%r", intent, message)
     prompt = _load_prompt(message, intent)
-    raw_response = _call_nova(prompt)
-    logger.debug("Raw Nova response: %r", raw_response)
-    result = _parse_extraction_response(raw_response, intent)
-    logger.info("Extracted fields: %s", result)
+    raw    = _call_model(prompt)
+    result = _parse_extraction_response(raw, intent)
+    logger.info("Extracted: %s", result)
     return result
 
 
 def _null_result(intent: str) -> dict:
-    """Return a fully-null result for the given intent."""
-    intent = intent.lower().strip() if intent in VALID_INTENTS else "other"
+    intent        = intent.lower().strip() if intent in VALID_INTENTS else "other"
     schema_fields = INTENT_SCHEMA.get(intent, INTENT_SCHEMA["other"])
-    result = {field: None for field in schema_fields}
+    result        = {field: None for field in schema_fields}
     result["intent"] = intent
-    return result
+    return result

agent/intent_agent.py

@@ -1,90 +1,102 @@
 """
-Intent detection agent for Notiflow.
-"""
+intent_agent.py
+---------------
+Stage 2: Intent Detection Agent for Notiflow
+
+Uses ModelRouter (Nova primary → Gemini fallback) to classify
+the business intent of Hinglish messages.
 
-from __future__ import annotations
+Supported intents:
+    order | payment | credit | return | preparation | other
+
+Integration note (backend upgrade):
+    The private _call_model() function now delegates to agent/model_router.py
+    instead of calling bedrock_client directly. All parsing and public API
+    logic is unchanged.
+"""
 
 import json
 import logging
 import re
 from pathlib import Path
 
-from botocore.exceptions import BotoCoreError, ClientError
-
-from app.bedrock_client import get_bedrock_client
-from app.config import MODEL_ID
-
 PROMPT_PATH = Path(__file__).parent.parent / "prompts" / "intent_prompt.txt"
 VALID_INTENTS = {"order", "payment", "credit", "return", "preparation", "other"}
 
 logger = logging.getLogger(__name__)
 
 
-def _get_bedrock_client():
-    """Return a cached Bedrock runtime client."""
-    return get_bedrock_client()
-
+# ---------------------------------------------------------------------------
+# Prompt loader
+# ---------------------------------------------------------------------------
 
 def _load_prompt(message: str) -> str:
-    """Load the intent prompt template and inject the user message."""
     template = PROMPT_PATH.read_text(encoding="utf-8")
     return template.replace("{message}", message.strip())
 
 
-def _call_nova(prompt: str) -> str:
-    """Send a prompt to Amazon Nova 2 Lite via Bedrock Converse API."""
-    client = _get_bedrock_client()
-    request_body = {
-        "messages": [{"role": "user", "content": [{"text": prompt}]}],
-        "inferenceConfig": {
-            "maxTokens": 64,
-            "temperature": 0.0,
-            "topP": 1.0,
-        },
-    }
+# ---------------------------------------------------------------------------
+# Model inference  (now via ModelRouter — Nova primary, Gemini fallback)
+# ---------------------------------------------------------------------------
+
+def _call_model(prompt: str) -> str:
+    """
+    Route the prompt through ModelRouter.
+    Returns raw text response from whichever model was available.
+    """
+    from agent.model_router import route
+    raw, model_used = route(prompt, max_tokens=64)
+    logger.info("Intent inference served by: %s", model_used)
+    return raw
 
-    try:
-        response = client.converse(modelId=MODEL_ID, **request_body)
-        output_message = response["output"]["message"]
-        text_parts = [
-            block["text"]
-            for block in output_message["content"]
-            if block.get("type") == "text" or "text" in block
-        ]
-        return " ".join(text_parts).strip()
-    except (BotoCoreError, ClientError) as exc:
-        logger.error("Bedrock API error: %s", exc)
-        raise RuntimeError(f"Failed to call Amazon Nova: {exc}") from exc
-
-
-def _parse_intent_response(raw: str) -> dict[str, str]:
-    """Parse model output into a validated intent dict."""
-    cleaned = re.sub(r"```(?:json)?|```", "", raw).strip()
 
+# ---------------------------------------------------------------------------
+# Response parser
+# ---------------------------------------------------------------------------
+
+def _parse_intent_response(raw: str) -> dict:
+    cleaned = re.sub(r"```(?:json)?|```", "", raw).strip()
     try:
         result = json.loads(cleaned)
         intent = result.get("intent", "other").lower().strip()
     except json.JSONDecodeError:
-        match = re.search(r'"intent"\s*:\s*"(\w+)"', cleaned)
+        match  = re.search(r'"intent"\s*:\s*"(\w+)"', cleaned)
         intent = match.group(1).lower() if match else "other"
 
     if intent not in VALID_INTENTS:
-        logger.warning("Model returned unknown intent '%s', defaulting to 'other'", intent)
+        logger.warning("Unknown intent '%s', defaulting to 'other'", intent)
         intent = "other"
 
     return {"intent": intent}
 
 
-def detect_intent(message: str) -> dict[str, str]:
-    """Detect the business intent of a Hinglish message."""
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def detect_intent(message: str) -> dict:
+    """
+    Detect the business intent of a Hinglish message.
+
+    Args:
+        message: Raw business message (Hinglish or English).
+
+    Returns:
+        {"intent": "<intent_string>"}
+
+    Examples:
+        >>> detect_intent("bhaiya 2 kilo bhej dena")
+        {'intent': 'order'}
+        >>> detect_intent("rahul ne 15000 bheja")
+        {'intent': 'payment'}
+    """
     if not message or not message.strip():
         logger.warning("Empty message received, returning 'other'")
         return {"intent": "other"}
 
-    logger.info("Detecting intent for message: %r", message)
-    prompt = _load_prompt(message)
-    raw_response = _call_nova(prompt)
-    logger.debug("Raw Nova response: %r", raw_response)
-    result = _parse_intent_response(raw_response)
+    logger.info("Detecting intent for: %r", message)
+    prompt      = _load_prompt(message)
+    raw         = _call_model(prompt)
+    result      = _parse_intent_response(raw)
     logger.info("Detected intent: %s", result["intent"])
-    return result
+    return result

agent/model_router.py

@@ -0,0 +1,81 @@
+"""
+agent/model_router.py
+---------------------
+ModelRouter for Notiflow.
+
+Acts as the unified inference interface used by intent_agent and
+extraction_agent instead of calling bedrock_client directly.
+
+Routing strategy:
+    Primary   → Amazon Nova Lite (app/bedrock_client.py :: call_nova)
+    Fallback  → Google Gemini   (models/gemini_client.py :: generate)
+
+Fallback triggers:
+    - AWS credentials missing / not configured
+    - Bedrock API error (any exception from call_nova)
+    - Model invocation failure
+    - Timeout (boto3 raises an exception on timeout)
+
+Both Nova and Gemini receive the SAME prompt. The prompts already
+instruct the model to return JSON — the router returns the raw text
+and lets the calling agent parse it.
+
+Public API
+----------
+route(prompt: str, max_tokens: int = 256) -> tuple[str, str]
+    Returns (raw_text_response, model_used)
+    model_used is "nova" or "gemini"
+"""
+
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def route(prompt: str, max_tokens: int = 256) -> tuple[str, str]:
+    """
+    Send a prompt to the best available model.
+
+    Tries Nova first; falls back to Gemini on any failure.
+
+    Args:
+        prompt:     Fully rendered prompt string.
+        max_tokens: Max tokens for Nova (Gemini ignores this param).
+
+    Returns:
+        Tuple of (response_text, model_used) where model_used is
+        "nova" or "gemini".
+
+    Raises:
+        RuntimeError: If BOTH Nova and Gemini fail.
+    """
+    # ── Attempt Nova ─────────────────────────────────────────────────────────
+    _nova_error: Exception | None = None
+    try:
+        from app.bedrock_client import call_nova
+        response = call_nova(prompt, max_tokens=max_tokens)
+        logger.info("ModelRouter → nova (success)")
+        return response, "nova"
+
+    except Exception as nova_exc:
+        _nova_error = nova_exc          # save before Python deletes `nova_exc`
+        logger.warning(
+            "Nova unavailable (%s) — falling back to Gemini.", nova_exc
+        )
+
+    # ── Fallback to Gemini ────────────────────────────────────────────────────
+    try:
+        from models.gemini_client import generate
+        response = generate(prompt)
+        logger.info("ModelRouter → gemini (fallback)")
+        return response, "gemini"
+
+    except Exception as gemini_exc:
+        logger.error("Gemini fallback also failed: %s", gemini_exc)
+        raise RuntimeError(
+            f"Both Nova and Gemini are unavailable.\n"
+            f"  Nova error:   {_nova_error}\n"
+            f"  Gemini error: {gemini_exc}"
+        ) from gemini_exc

api/notification_routes.py

@@ -0,0 +1,254 @@
+"""
+api/notification_routes.py
+--------------------------
+FastAPI router for Notiflow notification endpoints.
+
+Endpoints
+---------
+POST /api/notification
+    Receives a notification, runs the full agent pipeline,
+    returns the structured orchestrator result.
+
+GET  /api/notifications/generate
+    Calls Gemini to generate a batch of demo notifications.
+    Query param: n (default 5)
+
+WebSocket /ws/notifications
+    Streams live notifications to connected clients.
+    Accepts both frontend-pushed and Gemini-generated events.
+    Broadcasts to all connected clients.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query, HTTPException
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+# ---------------------------------------------------------------------------
+# Pydantic models
+# ---------------------------------------------------------------------------
+
+class NotificationRequest(BaseModel):
+    """Incoming notification payload."""
+    source:  str          # e.g. "whatsapp", "amazon", "payment", "return"
+    message: str          # Raw Hinglish business message
+
+
+class NotificationResponse(BaseModel):
+    """
+    Full orchestrator result — same shape as run_notiflow() output.
+    Frontend reads fields from data{} and event{}.
+    """
+    message: str
+    intent:  str
+    data:    dict[str, Any]
+    event:   dict[str, Any]
+    source:  str           # Echo back the notification source
+    model:   str | None = None   # "nova" | "gemini" | "demo"
+
+
+# ---------------------------------------------------------------------------
+# WebSocket connection manager
+# ---------------------------------------------------------------------------
+
+class _ConnectionManager:
+    """Simple in-memory broadcast manager for WebSocket clients."""
+
+    def __init__(self):
+        self._active: list[WebSocket] = []
+
+    async def connect(self, ws: WebSocket) -> None:
+        await ws.accept()
+        self._active.append(ws)
+        logger.info("WS client connected. Total: %d", len(self._active))
+
+    def disconnect(self, ws: WebSocket) -> None:
+        self._active = [c for c in self._active if c is not ws]
+        logger.info("WS client disconnected. Total: %d", len(self._active))
+
+    async def broadcast(self, payload: dict) -> None:
+        """Send JSON payload to all connected WebSocket clients."""
+        dead = []
+        for ws in self._active:
+            try:
+                await ws.send_json(payload)
+            except Exception:
+                dead.append(ws)
+        for ws in dead:
+            self.disconnect(ws)
+
+
+_manager = _ConnectionManager()
+
+
+# ---------------------------------------------------------------------------
+# POST /api/notification
+# ---------------------------------------------------------------------------
+
+@router.post("/api/notification", response_model=NotificationResponse)
+async def process_notification(body: NotificationRequest):
+    """
+    Receive a business notification and run the full Notiflow pipeline.
+
+    The endpoint calls run_notiflow(message) from app/main.py — it does
+    not call agents or skills directly.
+
+    After processing, the result is broadcast to all connected WebSocket
+    clients so the live stream panel updates in real time.
+
+    Args:
+        body: {"source": "whatsapp", "message": "bhaiya 3 kurti bhej dena"}
+
+    Returns:
+        Full orchestrator result + source echo.
+    """
+    if not body.message or not body.message.strip():
+        raise HTTPException(status_code=422, detail="Message cannot be empty.")
+
+    logger.info("POST /api/notification | source=%s | msg=%r", body.source, body.message)
+
+    try:
+        from app.main import run_notiflow
+        result = run_notiflow(body.message.strip())
+    except Exception as exc:
+        logger.error("Pipeline error: %s", exc)
+        raise HTTPException(status_code=500, detail=f"Pipeline error: {exc}")
+
+    response = NotificationResponse(
+        message = result["message"],
+        intent  = result["intent"],
+        data    = result["data"],
+        event   = result["event"],
+        source  = body.source,
+        model   = "demo" if not result["event"].get("event", "").endswith("_recorded")
+                         and not result["event"].get("event", "").endswith("_received")
+                         and not result["event"].get("event", "").endswith("_requested")
+                         and not result["event"].get("event", "").endswith("_queued")
+                  else "live",
+    )
+
+    # Broadcast to WebSocket clients
+    await _manager.broadcast(response.model_dump())
+
+    return response
+
+
+# ---------------------------------------------------------------------------
+# GET /api/notifications/generate
+# ---------------------------------------------------------------------------
+
+@router.get("/api/notifications/generate")
+async def generate_demo_notifications(n: int = Query(default=5, ge=1, le=20)):
+    """
+    Generate n demo notifications using Gemini (or static fallback).
+
+    Query params:
+        n: number of notifications to generate (1-20, default 5)
+
+    Returns:
+        {"notifications": [{"source": str, "message": str}, ...]}
+    """
+    from services.notification_generator import get_notifications
+    notifications = get_notifications(n)
+    return {"notifications": notifications}
+
+
+# ---------------------------------------------------------------------------
+# WebSocket /ws/notifications
+# ---------------------------------------------------------------------------
+
+@router.websocket("/ws/notifications")
+async def websocket_notification_stream(websocket: WebSocket):
+    """
+    WebSocket endpoint for real-time notification streaming.
+
+    Clients connect and receive:
+      - Notifications pushed by the frontend simulation
+      - Notifications generated by Gemini automation
+      - Results of processed notifications (broadcast from POST endpoint)
+
+    The client can also SEND a notification over the WebSocket:
+        {"source": "whatsapp", "message": "bhaiya 3 kurti bhej dena"}
+
+    The server will process it through the pipeline and broadcast the
+    result to all connected clients.
+
+    Protocol:
+        Client → Server:  {"source": str, "message": str}
+        Server → Client:  NotificationResponse JSON
+    """
+    await _manager.connect(websocket)
+    try:
+        while True:
+            raw = await websocket.receive_text()
+            try:
+                payload = json.loads(raw)
+                source  = payload.get("source", "websocket")
+                message = payload.get("message", "").strip()
+
+                if not message:
+                    await websocket.send_json({"error": "Empty message"})
+                    continue
+
+                from app.main import run_notiflow
+                result = run_notiflow(message)
+
+                response = {
+                    "message": result["message"],
+                    "intent":  result["intent"],
+                    "data":    result["data"],
+                    "event":   result["event"],
+                    "source":  source,
+                }
+                await _manager.broadcast(response)
+
+            except json.JSONDecodeError:
+                await websocket.send_json({"error": "Invalid JSON payload"})
+            except Exception as exc:
+                logger.error("WS pipeline error: %s", exc)
+                await websocket.send_json({"error": str(exc)})
+
+    except WebSocketDisconnect:
+        _manager.disconnect(websocket)
+
+
+# ---------------------------------------------------------------------------
+# GET /api/stream/start
+# ---------------------------------------------------------------------------
+
+@router.get("/api/stream/start")
+async def start_gemini_stream(
+    n:     int   = Query(default=5,   ge=1,  le=20),
+    delay: float = Query(default=2.0, ge=0.5, le=30.0),
+):
+    """
+    Trigger Gemini to generate n notifications and stream them to all
+    connected WebSocket clients with a delay between each.
+
+    Query params:
+        n:     number of notifications (default 5)
+        delay: seconds between each broadcast (default 2.0)
+
+    This runs in the background — the HTTP response returns immediately.
+    """
+    async def _stream():
+        from services.notification_generator import stream_notifications
+        async for notification in stream_notifications(n=n, delay_seconds=delay):
+            await _manager.broadcast({
+                "type":    "incoming_notification",
+                "source":  notification["source"],
+                "message": notification["message"],
+            })
+
+    asyncio.create_task(_stream())
+    return {"status": "streaming started", "n": n, "delay_seconds": delay}

app/config.py

@@ -31,7 +31,10 @@ REGISTRY_FILE = ROOT / "skills" / "skill_registry.json"  # Skill registry
 # When True the dashboard and main.py simulate the pipeline locally.
 # Set to False (or override via env var) to use real Bedrock inference.
 import os
-DEMO_MODE: bool = os.getenv("NOTIFLOW_DEMO_MODE", "true").lower() != "false"
+NOTIFLOW_DEMO_MODE = os.getenv("NOTIFLOW_DEMO_MODE", "true").lower() == "true"
+
+# Legacy alias for backward compatibility
+DEMO_MODE = NOTIFLOW_DEMO_MODE
 
 # ---------------------------------------------------------------------------
 # Amazon Bedrock settings
@@ -44,4 +47,19 @@ BEDROCK_MODEL_ID = "amazon.nova-lite-v1:0"
 # Ensure data directory exists at import time
 # ---------------------------------------------------------------------------
 
-DATA_DIR.mkdir(parents=True, exist_ok=True)
+DATA_DIR.mkdir(parents=True, exist_ok=True)
+
+# ---------------------------------------------------------------------------
+# Gemini settings  (added: backend upgrade)
+# ---------------------------------------------------------------------------
+
+GEMINI_API_KEY: str | None = os.getenv("GEMINI_API_KEY")
+
+# ---------------------------------------------------------------------------
+# Excel sync path  (added: backend upgrade)
+# If EXCEL_FILE_PATH is set in .env, it overrides the default DATA_FILE.
+# If not set, the existing DATA_FILE path is used as fallback.
+# ---------------------------------------------------------------------------
+
+_env_excel = os.getenv("EXCEL_FILE_PATH")
+EXCEL_SYNC_FILE = Path(_env_excel) if _env_excel else DATA_FILE

backend/main.py

@@ -0,0 +1,142 @@
+"""
+backend/main.py
+---------------
+FastAPI application entry point for Notiflow.
+
+Startup:
+  1. Loads .env file from project root (python-dotenv)
+  2. Configures logging
+  3. Mounts the notification API router
+  4. Exposes health check endpoint
+
+Run:
+    uvicorn backend.main:app --reload
+
+The server expects to be started from the project root so that
+all relative imports (app/, agent/, services/, etc.) resolve correctly.
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+
+# ---------------------------------------------------------------------------
+# .env loader — must happen BEFORE any app.config import
+# ---------------------------------------------------------------------------
+
+try:
+    from dotenv import load_dotenv
+    _env_path = Path(__file__).parent.parent / ".env"
+    load_dotenv(dotenv_path=_env_path, override=False)
+except ImportError:
+    pass   # python-dotenv not installed; env vars must be set externally
+
+# ---------------------------------------------------------------------------
+# Ensure project root is on sys.path when running via uvicorn
+# ---------------------------------------------------------------------------
+
+_PROJECT_ROOT = str(Path(__file__).parent.parent)
+if _PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, _PROJECT_ROOT)
+
+# ---------------------------------------------------------------------------
+# Logging
+# ---------------------------------------------------------------------------
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s  %(levelname)-8s  %(name)s — %(message)s",
+    datefmt="%H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# FastAPI app
+# ---------------------------------------------------------------------------
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from api.notification_routes import router as notification_router
+
+app = FastAPI(
+    title       = "Notiflow API",
+    description = (
+        "AI operations assistant for small businesses.\n\n"
+        "Converts informal Hinglish business notifications into "
+        "structured operations — powered by Amazon Nova 2 Lite."
+    ),
+    version = "1.0.0",
+    docs_url    = "/docs",
+    redoc_url   = "/redoc",
+)
+
+# ---------------------------------------------------------------------------
+# CORS — allow all origins for hackathon demo
+# ---------------------------------------------------------------------------
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins     = ["*"],
+    allow_credentials = True,
+    allow_methods     = ["*"],
+    allow_headers     = ["*"],
+)
+
+# ---------------------------------------------------------------------------
+# Routers
+# ---------------------------------------------------------------------------
+
+app.include_router(notification_router)
+
+# ---------------------------------------------------------------------------
+# Health check
+# ---------------------------------------------------------------------------
+
+@app.get("/health", tags=["Meta"])
+async def health():
+    """
+    Basic health check.
+
+    Returns service status and configuration summary.
+    """
+    from app.config import DEMO_MODE, BEDROCK_MODEL_ID, GEMINI_API_KEY, EXCEL_SYNC_FILE
+    return {
+        "status":          "ok",
+        "demo_mode":       DEMO_MODE,
+        "bedrock_model":   BEDROCK_MODEL_ID,
+        "gemini_enabled":  bool(GEMINI_API_KEY),
+        "excel_file":      str(EXCEL_SYNC_FILE),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Root redirect to docs
+# ---------------------------------------------------------------------------
+
+@app.get("/", include_in_schema=False)
+async def root():
+    from fastapi.responses import RedirectResponse
+    return RedirectResponse(url="/docs")
+
+
+# ---------------------------------------------------------------------------
+# Startup / shutdown events
+# ---------------------------------------------------------------------------
+
+@app.on_event("startup")
+async def on_startup():
+    from app.config import DEMO_MODE, GEMINI_API_KEY, EXCEL_SYNC_FILE
+    logger.info("=" * 55)
+    logger.info("  Notiflow API starting up")
+    logger.info("  Demo mode   : %s", DEMO_MODE)
+    logger.info("  Gemini      : %s", "enabled" if GEMINI_API_KEY else "disabled (no API key)")
+    logger.info("  Excel file  : %s", EXCEL_SYNC_FILE)
+    logger.info("=" * 55)
+
+
+@app.on_event("shutdown")
+async def on_shutdown():
+    logger.info("Notiflow API shutting down.")

backend/requirements.txt

@@ -0,0 +1,23 @@
+# Notiflow — backend dependencies
+# Install: pip install -r backend/requirements.txt
+
+# FastAPI stack
+fastapi>=0.111.0
+uvicorn[standard]>=0.29.0
+python-multipart>=0.0.9      # required for FastAPI form handling
+
+# Environment
+python-dotenv>=1.0.0
+
+# Amazon Bedrock
+boto3>=1.34.0
+
+# Gemini
+google-generativeai>=0.7.0
+
+# Excel
+pandas>=2.2.0
+openpyxl>=3.1.0
+
+# Pydantic (FastAPI uses v2)
+pydantic>=2.7.0

dashboard/.env.local

@@ -0,0 +1 @@
+VITE_API_URL=http://localhost:8000

dashboard/src/api.ts

@@ -0,0 +1,63 @@
+/**
+ * API service for communicating with FastAPI backend
+ * Running on http://localhost:8000
+ */
+
+const API_BASE_URL = import.meta.env.VITE_API_URL || 'http://localhost:8000';
+
+export interface NotificationPayload {
+  source: string;
+  message: string;
+}
+
+export interface ApiResponse {
+  intent: string;
+  data: Record<string, any>;
+  event: Record<string, any>;
+  message: string;
+}
+
+class ApiClient {
+  /**
+   * Send notification to backend
+   * POST /api/notification
+   */
+  async sendNotification(payload: NotificationPayload): Promise<ApiResponse> {
+    try {
+      const response = await fetch(`${API_BASE_URL}/api/notification`, {
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+        },
+        body: JSON.stringify(payload),
+        timeout: 30000,
+      });
+
+      if (!response.ok) {
+        throw new Error(`Backend error: ${response.statusText}`);
+      }
+
+      return await response.json();
+    } catch (error) {
+      console.error('API Error:', error);
+      throw error;
+    }
+  }
+
+  /**
+   * Health check endpoint
+   * GET /health
+   */
+  async healthCheck(): Promise<boolean> {
+    try {
+      const response = await fetch(`${API_BASE_URL}/health`, {
+        method: 'GET',
+      });
+      return response.ok;
+    } catch {
+      return false;
+    }
+  }
+}
+
+export const apiClient = new ApiClient();

dashboard/src/main.tsx

@@ -1,10 +1,10 @@
-import { StrictMode } from 'react'
 import { createRoot } from 'react-dom/client'
 import './index.css'
 import App from './App.tsx'
 
+// StrictMode removed: React 18 StrictMode intentionally remounts components
+// in dev (mount → unmount → remount), which resets useRef values and
+// double-fires effects — causing duplicate real API calls to the backend.
 createRoot(document.getElementById('root')!).render(
-  <StrictMode>
-    <App />
-  </StrictMode>,
+  <App />
 )

dashboard/src/react.tsx

@@ -1,4 +1,4 @@
-import React, { useState, useEffect, useCallback, useRef } from 'react';
+import React, { useState, useRef } from 'react';
 import { 
   MessageSquare, 
   ShoppingCart, 
@@ -19,6 +19,7 @@ import {
   Server,
   Terminal
 } from 'lucide-react';
+import { apiClient } from './api';
 
 interface Entity {
   item?: string;
@@ -48,10 +49,12 @@ interface SheetsState {
 }
 
 const VyaparFlow = () => {
-  // --- Simulation State ---
+  // --- State Management ---
   const [notifications, setNotifications] = useState<AppEvent[]>([]);
   const [activeEvent, setActiveEvent] = useState<AppEvent | null>(null);
-  const [processingStep, setProcessingStep] = useState(0); // 0: idle, 1: API Request, 2: Reasoning, 3: Completed
+  const [backendResult, setBackendResult] = useState<any>(null);
+  const [processingStep, setProcessingStep] = useState(0); // 0: idle, 1: API Request, 2: Reasoning, 3: Completed, 4: Error
+  const [pipelineError, setPipelineError] = useState<string | null>(null);
   const [view, setView] = useState('dashboard');
   const [activeSheet, setActiveSheet] = useState('Orders');
   
@@ -74,107 +77,157 @@ const VyaparFlow = () => {
     ]
   });
 
-  // Template for events
-  const eventTemplates: AppEvent[] = [
-    {
-      source: 'WhatsApp',
-      icon: <MessageSquare size={14} className="text-emerald-400" />,
-      title: 'Harish (Customer)',
-      message: 'bhaiya 3 kilo aata aur bhej dena kal tak',
-      type: 'order',
-      entity: { item: 'Aata', qty: '3kg', customer: 'Harish' }
-    },
-    {
-      source: 'Amazon Seller',
-      icon: <ShoppingCart size={14} className="text-orange-400" />,
-      title: 'New Order Received',
-      message: 'Product: Silk Kurti (Red). Qty: 1. Total: ₹1850',
-      type: 'marketplace',
-      entity: { item: 'Silk Kurti', qty: '1', customer: 'Amzn_User_44' }
-    },
-    {
-      source: 'GPay',
-      icon: <IndianRupee size={14} className="text-blue-400" />,
-      title: 'Payment Confirmed',
-      message: '₹4,500 received from Amit Kumar.',
-      type: 'payment',
-      entity: { amount: '₹4,500', customer: 'Amit Kumar' }
+  // --- Generate Random Notification ---
+  
+  const products = ["Silk Kurti", "Cotton Saree", "Jeans", "T-shirt", "Leather Jacket"];
+  const groceryItems = ["aata", "chawal", "dal", "cheeni", "tel"];
+  const customers = ["Rahul", "Amit Kumar", "Suresh", "Pooja", "Neha"];
+  const sources = ["whatsapp", "gpay", "amazon seller"];
+
+  const generateNotification = () => {
+    const source = sources[Math.floor(Math.random() * sources.length)];
+
+    if (source === "whatsapp") {
+      const qty = Math.floor(Math.random() * 5) + 1;
+      const item = groceryItems[Math.floor(Math.random() * groceryItems.length)];
+      return {
+        source: "whatsapp",
+        message: `bhaiya ${qty} kilo ${item} bhej dena kal tak`
+      };
     }
-  ];
 
-  // --- Automated Logic ---
-  
-  const triggerBackendProcess = useCallback(async (event: AppEvent) => {
-    if (isProcessingRef.current) return;
-    isProcessingRef.current = true;
-    
-    setActiveEvent(event);
-    setProcessingStep(1); // Stage 1: API Endpoint hit
+    if (source === "gpay") {
+      const amount = Math.floor(Math.random() * 15000) + 500;
+      const customer = customers[Math.floor(Math.random() * customers.length)];
+      return {
+        source: "gpay",
+        message: `₹${amount} received from ${customer}`
+      };
+    }
 
-    // Simulate Network Latency to "POST /api/v1/process-event"
-    await new Promise(r => setTimeout(r, 1500));
-    setProcessingStep(2); // Stage 2: Reasoning Phase (Nova 2 Lite)
+    const product = products[Math.floor(Math.random() * products.length)];
+    const qty = Math.floor(Math.random() * 3) + 1;
+    const total = Math.floor(Math.random() * 3000) + 500;
 
-    // Simulate AI Reasoning duration
-    await new Promise(r => setTimeout(r, 2500));
-    setProcessingStep(3); // Stage 3: Skill Execution / Record Completion
+    return {
+      source: "amazon seller",
+      message: `Product: ${product}. Qty: ${qty}. Total: ₹${total}`
+    };
+  };
 
-    // Update Sheets with the new data
-    setSheets(prev => {
-        const newSheets = { ...prev };
-        const rowId = `RECORD-${Date.now()}`;
-        if (event.type === 'payment') {
-            newSheets.Ledger = [{ 
-                id: rowId,
-                customer: event.entity.customer || 'Unknown', 
-                type: 'Credit', 
-                amount: event.entity.amount || '0', 
-                date: 'Today', 
-                status: 'Settled' 
-            }, ...prev.Ledger].slice(0, 10);
-        } else {
-            newSheets.Orders = [{ 
-                id: `ORD-${Math.floor(Math.random()*900 + 100)}`, 
-                source: event.source, 
-                customer: event.entity.customer || 'Unknown', 
-                item: event.entity.item || 'Item', 
-                qty: event.entity.qty || '1', 
-                status: 'Confirmed' 
-            }, ...prev.Orders].slice(0, 10);
-        }
-        return newSheets;
-    });
-
-    // Reset loop readiness
-    setTimeout(() => {
-        setProcessingStep(0);
-        isProcessingRef.current = false;
-    }, 4000);
-  }, []);
-
-  // Notification Ingestion Loop
-  useEffect(() => {
-    let index = 0;
-    const interval = setInterval(() => {
-      if (isProcessingRef.current) return;
+  const handleGenerateNotification = async () => {
+    // Prevent double-click or multiple rapid requests
+    if (isProcessingRef.current) return;
+    isProcessingRef.current = true;
 
-      const template = eventTemplates[index];
-      const uniqueId = `event-${Date.now()}-${Math.floor(Math.random() * 1000)}`;
+    try {
+      // Step 1: Generate random notification
+      const notification = generateNotification();
       
-      const newEvent = {
-        ...template,
+      // Step 2: Map source to UI properties
+      const sourceMap: { [key: string]: { icon: React.ReactNode; title: string } } = {
+        'whatsapp': { 
+          icon: <MessageSquare size={14} className="text-emerald-400" />, 
+          title: 'WhatsApp Order' 
+        },
+        'gpay': { 
+          icon: <IndianRupee size={14} className="text-blue-400" />, 
+          title: 'GPay Payment' 
+        },
+        'amazon seller': { 
+          icon: <ShoppingCart size={14} className="text-orange-400" />, 
+          title: 'Amazon Order' 
+        },
+      };
+      
+      const sourceProps = sourceMap[notification.source] || { 
+        icon: <MessageSquare size={14} className="text-slate-400" />, 
+        title: 'New Notification' 
+      };
+      
+      // Step 3: Create AppEvent object
+      const uniqueId = `event-${Date.now()}-${Math.floor(Math.random() * 1000)}`;
+      const newEvent: AppEvent = {
         id: uniqueId,
+        source: notification.source.charAt(0).toUpperCase() + notification.source.slice(1),
+        icon: sourceProps.icon,
+        title: sourceProps.title,
+        message: notification.message,
+        type: notification.source === 'gpay' ? 'payment' : 'order',
+        entity: {},
         timestamp: new Date().toLocaleTimeString()
       };
-
+      
+      // Step 4: Add to UI notifications list
       setNotifications(prev => [newEvent, ...prev].slice(0, 5));
-      triggerBackendProcess(newEvent);
+      setActiveEvent(newEvent);
+      setBackendResult(null);
+      setProcessingStep(1); // Stage 1: API Endpoint hit
+
+      // Step 5: SINGLE API CALL (the ONLY place where backend is triggered)
+      const response = await apiClient.sendNotification({
+        source: notification.source.toLowerCase(),
+        message: notification.message
+      });
+
+      // Step 6: Store backend response
+      setBackendResult(response);
+      setProcessingStep(2); // Stage 2: Reasoning Phase (Nova 2 Lite)
       
-      index = (index + 1) % eventTemplates.length;
-    }, 12000); // 12-second cycle for clear visualization
+      // Simulate slight delay for visual feedback
+      await new Promise(r => setTimeout(r, 1000));
+      setProcessingStep(3); // Stage 3: Skill Execution / Record Completion
 
-    return () => clearInterval(interval);
-  }, [triggerBackendProcess]);
+      // Step 7: Update Sheets with the response data
+      setSheets(prev => {
+        const newSheets = { ...prev };
+        const rowId = `RECORD-${Date.now()}`;
+        
+        // Extract entity info from API response
+        const entityData = response.data || {};
+        
+        if (response.intent === 'payment') {
+          newSheets.Ledger = [{ 
+            id: rowId,
+            customer: entityData.customer || newEvent.entity.customer || 'Unknown', 
+            type: 'Credit', 
+            amount: entityData.amount || newEvent.entity.amount || '₹0', 
+            date: 'Today', 
+            status: 'Settled' 
+          }, ...prev.Ledger].slice(0, 10);
+        } else {
+          newSheets.Orders = [{ 
+            id: `ORD-${Math.floor(Math.random()*900 + 100)}`, 
+            source: newEvent.source, 
+            customer: entityData.customer || newEvent.entity.customer || 'Unknown', 
+            item: entityData.item || newEvent.entity.item || 'Item', 
+            qty: entityData.quantity || newEvent.entity.qty || '1', 
+            status: 'Confirmed' 
+          }, ...prev.Orders].slice(0, 10);
+        }
+        return newSheets;
+      });
+
+    } catch (error: any) {
+      console.error('Backend error:', error);
+      const msg = error?.message || 'Both models unavailable';
+      setPipelineError(
+        msg.includes('quota') || msg.includes('429')
+          ? 'Quota exceeded — wait ~1 min'
+          : msg.includes('unavailable') || msg.includes('500')
+          ? 'Models unavailable — retry shortly'
+          : 'Request failed'
+      );
+      setProcessingStep(4); // Error state — shows red button
+    } finally {
+      // Reset lock after 5s so user clearly sees the error before re-enabling
+      setTimeout(() => {
+        setProcessingStep(0);
+        setPipelineError(null);
+        isProcessingRef.current = false;
+      }, 5000);
+    }
+  };
 
   return (
     <div className="flex h-screen bg-[#050505] text-slate-300 font-sans selection:bg-indigo-500/30 overflow-hidden">
@@ -307,7 +360,26 @@ const VyaparFlow = () => {
                 </button>
             </nav>
             
-            <div className="flex items-center gap-4">
+            <div className="flex items-center gap-6">
+                <button
+                    onClick={handleGenerateNotification}
+                    disabled={processingStep > 0}
+                    className={`px-4 py-2 rounded-lg text-xs font-bold border transition-all ${
+                      processingStep === 4
+                        ? 'bg-red-900/30 text-red-400 border-red-500/40 cursor-not-allowed'
+                        : processingStep > 0
+                        ? 'bg-slate-800/50 text-slate-500 border-slate-700/50 cursor-not-allowed'
+                        : 'bg-emerald-600/20 text-emerald-400 border-emerald-500/30 hover:bg-emerald-600/30 hover:border-emerald-500/50'
+                    }`}
+                    title={pipelineError || ''}
+                >
+                    {processingStep === 4
+                      ? `❌ ${pipelineError || 'Failed'}`
+                      : processingStep > 0
+                      ? '⏳ Processing...'
+                      : '⚡ Generate Notification'}
+                </button>
+                
                 <div className="flex flex-col items-end">
                     <div className="flex items-center gap-2 text-[10px] font-black text-indigo-400 uppercase tracking-widest">
                         Nova-2-Lite <div className="w-1.5 h-1.5 bg-green-500 rounded-full animate-pulse"></div>
@@ -379,14 +451,14 @@ const VyaparFlow = () => {
                                         <div className="bg-black/60 p-5 rounded-2xl border border-white/5 space-y-4">
                                             <div>
                                                 <div className="text-[9px] text-slate-500 uppercase font-black mb-1.5 tracking-wider">Detected Intent</div>
-                                                <div className="text-xs text-indigo-400 font-bold capitalize">{activeEvent.type.replace('_', ' ')} Pattern Matched</div>
+                                                <div className="text-xs text-indigo-400 font-bold capitalize">{(backendResult?.intent || activeEvent?.type || 'unknown').replace('_', ' ')} Pattern Matched</div>
                                             </div>
                                             <div className="h-[1px] bg-white/5"></div>
                                             <div>
                                                 <div className="text-[9px] text-slate-500 uppercase font-black mb-2 tracking-wider">Extracted Entities</div>
                                                 <div className="flex flex-wrap gap-2">
-                                                    {Object.entries(activeEvent.entity).map(([k, v]) => (
-                                                        <span key={`${activeEvent.id}-entity-${k}`} className="text-[10px] px-2.5 py-1 rounded bg-indigo-500/10 text-indigo-300 border border-indigo-500/20 font-bold">{k}: {v}</span>
+                                                    {Object.entries(backendResult?.data || activeEvent?.entity || {}).map(([k, v]) => (
+                                                        <span key={`${activeEvent?.id}-entity-${k}`} className="text-[10px] px-2.5 py-1 rounded bg-indigo-500/10 text-indigo-300 border border-indigo-500/20 font-bold">{k}: {String(v)}</span>
                                                     ))}
                                                 </div>
                                             </div>

dashboard/streamlit_app.py

@@ -1,335 +0,0 @@
-"""
-streamlit_app.py
-----------------
-Stage 7 + FIX 3 + FIX 4: Streamlit Dashboard for Notiflow
-
-Changes from Stage 7:
-  FIX 3 — Dashboard now calls run_notiflow(message) from app/main.py only.
-           No direct imports of agents or orchestrator.
-  FIX 4 — File paths and DEMO_MODE come from app/config.py. No hardcoded paths.
-
-Run:
-    streamlit run dashboard/streamlit_app.py
-"""
-from __future__ import annotations
-
-import sys
-from pathlib import Path
-
-# Add project root to Python path
-ROOT = Path(__file__).resolve().parents[1]
-sys.path.append(str(ROOT))
-
-from pathlib import Path
-
-import pandas as pd
-import streamlit as st
-
-# ── Page config (must be first Streamlit call) ───────────────────────────────
-st.set_page_config(
-    page_title="Notiflow · AI Operations Dashboard",
-    page_icon="⚡",
-    layout="wide",
-    initial_sidebar_state="expanded",
-)
-
-# ── Config (FIX 4) ────────────────────────────────────────────────────────────
-from app.config import DATA_FILE, DEMO_MODE as _CONFIG_DEMO_MODE
-
-# ── Backend entry point (FIX 3) ───────────────────────────────────────────────
-from app.main import run_notiflow
-
-# ---------------------------------------------------------------------------
-# Static data
-# ---------------------------------------------------------------------------
-
-SAMPLE_MESSAGES = {
-    "— pick a sample message —":              "",
-    "💰  Payment — Rahul ₹15,000":            "rahul ne 15000 bheja",
-    "📦  Order — 3 kurties":                   "bhaiya 3 kurti bhej dena",
-    "📦  Order — Priya, 2 kg atta":            "priya ke liye 2 kilo aata bhej dena",
-    "🔄  Return — size issue":                 "size chota hai exchange karna hai",
-    "📒  Credit — simple udhar":               "udhar me de dijiye",
-    "📒  Credit — Suresh ₹500":               "suresh ko 500 ka maal udhar dena",
-    "🗂️  Prep — pack 3 kurties":              "3 kurti ka set ready rakhna",
-    "💰  Payment — Amit GPay ₹8,000":         "amit bhai ka 8000 gpay se aaya",
-}
-
-INTENT_CONFIG = {
-    "order":       {"emoji": "📦", "color": "#1E88E5", "label": "Order",       "bg": "#E3F2FD"},
-    "payment":     {"emoji": "💰", "color": "#43A047", "label": "Payment",     "bg": "#E8F5E9"},
-    "credit":      {"emoji": "📒", "color": "#FB8C00", "label": "Credit",      "bg": "#FFF3E0"},
-    "return":      {"emoji": "🔄", "color": "#E53935", "label": "Return",      "bg": "#FFEBEE"},
-    "preparation": {"emoji": "🗂️", "color": "#8E24AA", "label": "Preparation","bg": "#F3E5F5"},
-    "other":       {"emoji": "💬", "color": "#757575", "label": "Other",       "bg": "#F5F5F5"},
-}
-
-SHEETS      = ["Orders", "Ledger", "Returns", "Inventory", "Invoices"]
-SHEET_ICONS = {"Orders": "📦", "Ledger": "💰", "Returns": "🔄",
-               "Inventory": "📊", "Invoices": "🧾"}
-
-# ---------------------------------------------------------------------------
-# Pipeline trace builder
-# Constructs the step-by-step trace from the flat result dict.
-# run_notiflow() returns {message, intent, data, event} — no per-step data
-# in demo mode, so we reconstruct a display trace from the final result.
-# ---------------------------------------------------------------------------
-
-def _build_trace(result: dict) -> list[dict]:
-    return [
-        {
-            "icon":  "🧠",
-            "step":  "Intent Agent",
-            "label": f"Intent detected: **{result['intent']}**",
-            "note":  "Nova 2 Lite reads the Hinglish message and classifies its business intent.",
-            "output": {"intent": result["intent"]},
-        },
-        {
-            "icon":  "🔍",
-            "step":  "Extraction + Validation",
-            "label": "Structured fields extracted and validated",
-            "note":  "Nova 2 Lite extracts entities; validator normalises numbers, text and payment aliases.",
-            "output": result["data"],
-        },
-        {
-            "icon":  "⚙️",
-            "step":  "Skill Router",
-            "label": f"Skill executed: **{result['event'].get('event', '')}**",
-            "note":  "The router dispatches to the correct business skill, persists data to Excel.",
-            "output": result["event"],
-        },
-    ]
-
-# ---------------------------------------------------------------------------
-# UI helpers
-# ---------------------------------------------------------------------------
-
-def _intent_cfg(intent: str) -> dict:
-    return INTENT_CONFIG.get(intent, INTENT_CONFIG["other"])
-
-
-def _render_intent_badge(intent: str):
-    cfg = _intent_cfg(intent)
-    st.markdown(
-        f"<div style='"
-        f"display:inline-block; background:{cfg['bg']};"
-        f"border-left:5px solid {cfg['color']}; border-radius:6px;"
-        f"padding:8px 18px; font-weight:700; font-size:1.05rem;"
-        f"color:{cfg['color']}; margin-bottom:10px; letter-spacing:.04em;"
-        f"'>{cfg['emoji']}&nbsp;&nbsp;{cfg['label'].upper()}</div>",
-        unsafe_allow_html=True,
-    )
-
-
-def _render_trace(trace: list[dict]):
-    st.markdown("#### 🔄 Pipeline Trace")
-    for i, step in enumerate(trace):
-        with st.expander(f"{step['icon']}  Step {i+1} — {step['step']}", expanded=True):
-            st.markdown(step["label"])
-            st.caption(step["note"])
-            st.json(step["output"], expanded=False)
-
-
-def _render_result(result: dict):
-    intent = result.get("intent", "other")
-    st.markdown("#### 📋 Business Event")
-    _render_intent_badge(intent)
-
-    col1, col2 = st.columns(2, gap="medium")
-    with col1:
-        st.markdown("**Extracted Fields**")
-        data = result.get("data", {})
-        if data:
-            rows = [{"Field": k, "Value": str(v) if v is not None else "—"}
-                    for k, v in data.items()]
-            st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)
-        else:
-            st.info("No fields extracted.")
-    with col2:
-        st.markdown("**Skill Output**")
-        st.json(result.get("event", {}), expanded=True)
-
-
-def _render_table(sheet_name: str):
-    """Read sheet from Excel (FIX 4: path from config) and display it."""
-    if not Path(DATA_FILE).exists():
-        st.info(f"No data file found yet at `{DATA_FILE}`. Process a message to create it.", icon="📭")
-        return
-    try:
-        df = pd.read_excel(DATA_FILE, sheet_name=sheet_name)
-    except Exception:
-        df = pd.DataFrame()
-
-    if df.empty:
-        st.info(f"No records yet in **{sheet_name}**. Process a message to populate this sheet.", icon="📭")
-        return
-
-    st.dataframe(df, use_container_width=True, hide_index=True)
-    icon = SHEET_ICONS.get(sheet_name, "📄")
-    st.caption(f"{icon} {len(df)} record{'s' if len(df) != 1 else ''} in **{sheet_name}**")
-
-
-def _render_sidebar() -> tuple[str, bool]:
-    with st.sidebar:
-        st.markdown(
-            "<div style='text-align:center; padding:10px 0 4px'>"
-            "<span style='font-size:2.4rem'>⚡</span><br>"
-            "<span style='font-size:1.2rem; font-weight:700'>Notiflow</span><br>"
-            "<span style='font-size:.78rem; color:#888'>AI Operations Assistant</span>"
-            "</div>",
-            unsafe_allow_html=True,
-        )
-        st.divider()
-
-        # Demo toggle — default from config (FIX 4)
-        demo_mode = st.toggle(
-            "🧪 Demo Mode",
-            value=st.session_state.get("demo_mode", _CONFIG_DEMO_MODE),
-            help=(
-                "ON  → pipeline simulated locally, no AWS needed.\n"
-                "OFF → calls Amazon Nova 2 Lite via Bedrock (AWS creds required)."
-            ),
-        )
-        if demo_mode:
-            st.success("Demo mode active — no AWS needed", icon="✅")
-        else:
-            st.warning("Live mode — AWS credentials required", icon="⚠️")
-
-        st.divider()
-        st.markdown("**📂 Business Data**")
-        raw_choice = st.radio(
-            "sheet",
-            [f"{SHEET_ICONS[s]}  {s}" for s in SHEETS],
-            label_visibility="collapsed",
-        )
-        active_sheet = raw_choice.split("  ", 1)[1]
-
-        st.divider()
-        st.markdown(
-            "<div style='font-size:.78rem; color:#999'>"
-            "Built for the <b>Amazon Nova AI Hackathon</b>.<br>"
-            "Powered by <b>Amazon Nova 2 Lite</b> via Bedrock."
-            "</div>",
-            unsafe_allow_html=True,
-        )
-
-    return active_sheet, demo_mode
-
-# ---------------------------------------------------------------------------
-# Main
-# ---------------------------------------------------------------------------
-
-def main():
-    # Session state defaults
-    for key, default in {
-        "demo_mode":   _CONFIG_DEMO_MODE,
-        "last_result": None,
-        "last_trace":  None,
-        "last_error":  None,
-        "msg_input":   "",
-    }.items():
-        if key not in st.session_state:
-            st.session_state[key] = default
-
-    active_sheet, st.session_state.demo_mode = _render_sidebar()
-
-    # Header
-    st.markdown(
-        "<h1 style='margin-bottom:2px'>⚡ Notiflow "
-        "<span style='font-size:1rem; font-weight:400; color:#888'>"
-        "AI Operations Dashboard</span></h1>"
-        "<p style='color:#999; margin:0'>"
-        "Convert informal Hinglish business messages into structured operations "
-        "— powered by Amazon Nova&nbsp;2&nbsp;Lite</p>",
-        unsafe_allow_html=True,
-    )
-    st.divider()
-
-    left, right = st.columns([1, 1], gap="large")
-
-    # ── Left — input ─────────────────────────────────────────────────────────
-    with left:
-        st.markdown("### 💬 Enter Business Message")
-
-        sample_key = st.selectbox(
-            "Quick samples",
-            options=list(SAMPLE_MESSAGES.keys()),
-            index=0,
-            label_visibility="collapsed",
-        )
-        if SAMPLE_MESSAGES.get(sample_key):
-            st.session_state.msg_input = SAMPLE_MESSAGES[sample_key]
-
-        message = st.text_area(
-            "Message",
-            value=st.session_state.msg_input,
-            height=110,
-            placeholder='e.g. "rahul ne 15000 bheja"',
-            label_visibility="collapsed",
-        )
-        st.session_state.msg_input = message
-
-        btn_label = (
-            "🧪 Run Demo Pipeline"
-            if st.session_state.demo_mode
-            else "🚀 Run AI Pipeline (Nova)"
-        )
-        clicked = st.button(
-            btn_label,
-            type="primary",
-            use_container_width=True,
-            disabled=not message.strip(),
-        )
-
-        if clicked and message.strip():
-            st.session_state.last_result = None
-            st.session_state.last_trace  = None
-            st.session_state.last_error  = None
-
-            with st.spinner("Running agent pipeline…"):
-                try:
-                    # FIX 3: only call run_notiflow — no direct agent imports
-                    result = run_notiflow(
-                        message.strip(),
-                        demo_mode=st.session_state.demo_mode,
-                    )
-                    st.session_state.last_result = result
-                    st.session_state.last_trace  = _build_trace(result)
-                except Exception as exc:
-                    st.session_state.last_error = str(exc)
-
-        if st.session_state.last_error:
-            st.error(
-                f"**Pipeline error:** {st.session_state.last_error}\n\n"
-                "Tip: Enable **Demo Mode** in the sidebar to run without AWS credentials.",
-                icon="🚨",
-            )
-
-    # ── Right — output ────────────────────────────────────────────────────────
-    with right:
-        if st.session_state.last_result and st.session_state.last_trace:
-            _render_trace(st.session_state.last_trace)
-            st.divider()
-            _render_result(st.session_state.last_result)
-        else:
-            st.markdown("### 📊 Agent Output")
-            st.markdown(
-                "<div style='background:#F8F9FA; border-radius:12px;"
-                "padding:52px 30px; text-align:center; color:#aaa;'>"
-                "<div style='font-size:3rem'>⚡</div>"
-                "<div style='margin-top:10px; font-size:.95rem; line-height:1.6'>"
-                "Select a sample or type a message,<br>"
-                "then click <strong>Run Pipeline</strong>."
-                "</div></div>",
-                unsafe_allow_html=True,
-            )
-
-    # ── Data tables ───────────────────────────────────────────────────────────
-    st.divider()
-    icon = SHEET_ICONS.get(active_sheet, "📄")
-    st.markdown(f"### {icon} {active_sheet}")
-    _render_table(active_sheet)
-
-
-if __name__ == "__main__":
-    main()

data/agent_memory.json

@@ -0,0 +1,11 @@
+{
+  "recent_customers": [],
+  "recent_items": [
+    "t-shirt",
+    "dal",
+    "silk kurti (red)",
+    "silk kurti",
+    "tel",
+    "aata"
+  ]
+}

models/gemini_client.py

@@ -0,0 +1,184 @@
+"""
+models/gemini_client.py
+-----------------------
+Gemini API client for Notiflow.
+
+Serves two distinct roles:
+
+  1. FALLBACK REASONER — used by ModelRouter when Nova is unavailable.
+     generate(prompt) returns a plain-text response that must match
+     Nova's JSON output schema exactly (enforced by the prompt).
+
+  2. NOTIFICATION GENERATOR — used by notification_generator.py to
+     produce realistic Hinglish business notifications for demo mode.
+     generate_notifications(n) returns a list of notification dicts.
+
+Configuration
+-------------
+  GEMINI_API_KEY — from .env / environment variable
+
+Dependencies
+------------
+  pip install google-generativeai
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from typing import Any
+
+from app.config import GEMINI_API_KEY
+
+logger = logging.getLogger(__name__)
+
+# Gemini model to use — flash is fast and sufficient for this use case
+_GEMINI_MODEL = "gemini-2.5-flash"
+
+_client = None  # lazy singleton
+
+
+def _get_client():
+    """Return a cached Gemini GenerativeModel instance."""
+    global _client
+    if _client is not None:
+        return _client
+
+    if not GEMINI_API_KEY:
+        raise RuntimeError(
+            "GEMINI_API_KEY is not set. "
+            "Add it to your .env file: GEMINI_API_KEY=your_key_here"
+        )
+
+    try:
+        import google.generativeai as genai
+        genai.configure(api_key=GEMINI_API_KEY)
+        _client = genai.GenerativeModel(_GEMINI_MODEL)
+        logger.info("Gemini client initialised (model=%s)", _GEMINI_MODEL)
+        return _client
+    except ImportError:
+        raise RuntimeError(
+            "google-generativeai is not installed. "
+            "Run: pip install google-generativeai"
+        )
+    except Exception as exc:
+        raise RuntimeError(f"Failed to initialise Gemini client: {exc}") from exc
+
+
+def _strip_fences(text: str) -> str:
+    """Strip markdown code fences from model output."""
+    return re.sub(r"```(?:json)?|```", "", text).strip()
+
+
+# ---------------------------------------------------------------------------
+# Public API — Role 1: Fallback reasoner for ModelRouter
+# ---------------------------------------------------------------------------
+
+def generate(prompt: str) -> str:
+    """
+    Send a prompt to Gemini and return the raw text response.
+
+    Used by ModelRouter as a Nova fallback. The prompt already instructs
+    the model to return JSON — this function returns the raw string and
+    lets the caller parse it.
+
+    Args:
+        prompt: Fully rendered prompt (same prompt sent to Nova).
+
+    Returns:
+        Raw text response from Gemini (may contain JSON).
+
+    Raises:
+        RuntimeError: If the API call fails or client cannot be initialised.
+    """
+    client = _get_client()
+    try:
+        response = client.generate_content(prompt)
+        raw = response.text or ""
+        logger.debug("Gemini raw response: %r", raw[:200])
+        return _strip_fences(raw)
+    except Exception as exc:
+        logger.error("Gemini generate() failed: %s", exc)
+        raise RuntimeError(f"Gemini API error: {exc}") from exc
+
+
+# ---------------------------------------------------------------------------
+# Public API — Role 2: Notification generator
+# ---------------------------------------------------------------------------
+
+_NOTIFICATION_PROMPT = """
+You are simulating incoming business notifications for a small business in India.
+
+Generate {n} realistic business notifications in Hinglish (Hindi + English mix).
+
+Each notification must come from one of these sources:
+- whatsapp   (informal text from customers or suppliers)
+- payment    (UPI payment confirmation message)
+- amazon     (marketplace order or return notification)
+- return     (customer return or exchange request)
+
+Each notification must represent ONE of these business events:
+- An order for a product (item name + quantity)
+- A payment received (person name + amount)
+- A credit/udhar request
+- A return or exchange request
+- A preparation/packing request
+
+Rules:
+- Use natural Hinglish phrasing. Not too formal.
+- Vary sources and event types.
+- Include real-sounding names (Rahul, Priya, Suresh, Amit, etc.)
+- Include real product names (kurti, aata, daal, saree, etc.)
+- Do NOT include any explanation or preamble.
+- Return ONLY a valid JSON array, nothing else.
+
+Output format (JSON array only, no markdown):
+[
+  {{"source": "whatsapp", "message": "bhaiya 3 kurti bhej dena"}},
+  {{"source": "payment", "message": "Rahul ne 15000 bheja UPI se"}}
+]
+"""
+
+
+def generate_notifications(n: int = 5) -> list[dict[str, str]]:
+    """
+    Generate n realistic Hinglish business notifications via Gemini.
+
+    Args:
+        n: Number of notifications to generate (default 5).
+
+    Returns:
+        List of dicts with keys "source" and "message".
+        Returns an empty list if generation fails (non-fatal).
+
+    Example:
+        >>> generate_notifications(3)
+        [
+            {"source": "whatsapp", "message": "bhaiya 3 kurti bhej dena"},
+            {"source": "payment",  "message": "Rahul ne 15000 bheja"},
+            {"source": "return",   "message": "size chota hai exchange karna hai"},
+        ]
+    """
+    prompt = _NOTIFICATION_PROMPT.format(n=n)
+    try:
+        raw  = generate(prompt)
+        data = json.loads(raw)
+        if not isinstance(data, list):
+            logger.warning("Gemini notification response was not a list: %s", type(data))
+            return []
+        # Validate each entry has required keys
+        valid = [
+            item for item in data
+            if isinstance(item, dict)
+            and "source"  in item
+            and "message" in item
+        ]
+        logger.info("Generated %d notifications via Gemini", len(valid))
+        return valid
+    except json.JSONDecodeError as exc:
+        logger.warning("Could not parse Gemini notification output as JSON: %s", exc)
+        return []
+    except Exception as exc:
+        logger.warning("generate_notifications failed: %s", exc)
+        return []

package-lock.json

@@ -0,0 +1,6 @@
+{
+  "name": "Notiflow",
+  "lockfileVersion": 3,
+  "requires": true,
+  "packages": {}
+}

services/excel_sync.py

@@ -0,0 +1,210 @@
+"""
+services/excel_sync.py
+----------------------
+Semantic Excel sync wrapper for Notiflow.
+
+Wraps utils/excel_writer.append_row() with business-named functions so
+skills and the FastAPI backend can call append_order(), append_payment()
+etc. without knowing the sheet schema details.
+
+Excel writes always go to EXCEL_SYNC_FILE from app/config.py.
+If EXCEL_FILE_PATH env var is set, that path is used; otherwise falls
+back to the default DATA_FILE path (data/notiflow_data.xlsx).
+
+No Excel writing logic is duplicated here — this is a thin adapter only.
+
+Public API
+----------
+append_order(event_dict)    — writes to Orders sheet
+append_payment(event_dict)  — writes to Ledger sheet (type="payment")
+append_return(event_dict)   — writes to Returns sheet
+append_credit(event_dict)   — writes to Ledger sheet (type="credit")
+append_inventory(record)    — writes to Inventory sheet
+append_invoice(record)      — writes to Invoices sheet
+sync_from_event(result)     — auto-routes based on intent (convenience)
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+from app.config import EXCEL_SYNC_FILE
+from utils.excel_writer import append_row, EXCEL_FILE as _DEFAULT_FILE
+
+logger = logging.getLogger(__name__)
+
+
+def _active_file() -> Path:
+    """
+    Return the active Excel file path.
+    EXCEL_SYNC_FILE (from env EXCEL_FILE_PATH) takes priority over default.
+    """
+    return Path(EXCEL_SYNC_FILE)
+
+
+# ---------------------------------------------------------------------------
+# Sheet-specific append functions
+# ---------------------------------------------------------------------------
+
+def append_order(event_dict: dict) -> None:
+    """
+    Append an order record to the Orders sheet.
+
+    Args:
+        event_dict: The "order" sub-dict from an order skill event.
+                    Expected keys: order_id, timestamp, customer, item,
+                                   quantity, status
+    """
+    record = {
+        "order_id":  event_dict.get("order_id"),
+        "timestamp": event_dict.get("timestamp"),
+        "customer":  event_dict.get("customer"),
+        "item":      event_dict.get("item"),
+        "quantity":  event_dict.get("quantity"),
+        "status":    event_dict.get("status", "pending"),
+    }
+    append_row("Orders", record)
+    logger.info("Excel sync → Orders: %s", record.get("order_id"))
+
+
+def append_payment(event_dict: dict) -> None:
+    """
+    Append a payment record to the Ledger sheet.
+
+    Args:
+        event_dict: The "payment" sub-dict from a payment skill event.
+                    Expected keys: entry_id, timestamp, customer,
+                                   amount, payment_type, status
+    """
+    record = {
+        "entry_id":     event_dict.get("entry_id"),
+        "timestamp":    event_dict.get("timestamp"),
+        "type":         "payment",
+        "customer":     event_dict.get("customer"),
+        "item":         None,
+        "quantity":     None,
+        "amount":       event_dict.get("amount"),
+        "payment_type": event_dict.get("payment_type"),
+        "status":       event_dict.get("status", "received"),
+    }
+    append_row("Ledger", record)
+    logger.info("Excel sync → Ledger (payment): customer=%s", record.get("customer"))
+
+
+def append_return(event_dict: dict) -> None:
+    """
+    Append a return record to the Returns sheet.
+
+    Args:
+        event_dict: The "return" sub-dict from a return skill event.
+                    Expected keys: return_id, timestamp, customer,
+                                   item, reason, status
+    """
+    record = {
+        "return_id": event_dict.get("return_id"),
+        "timestamp": event_dict.get("timestamp"),
+        "customer":  event_dict.get("customer"),
+        "item":      event_dict.get("item"),
+        "reason":    event_dict.get("reason"),
+        "status":    event_dict.get("status", "pending_review"),
+    }
+    append_row("Returns", record)
+    logger.info("Excel sync → Returns: %s", record.get("return_id"))
+
+
+def append_credit(event_dict: dict) -> None:
+    """
+    Append a credit (udhar) record to the Ledger sheet.
+
+    Args:
+        event_dict: The "credit" sub-dict from a credit skill event.
+    """
+    record = {
+        "entry_id":     event_dict.get("entry_id"),
+        "timestamp":    event_dict.get("timestamp"),
+        "type":         "credit",
+        "customer":     event_dict.get("customer"),
+        "item":         event_dict.get("item"),
+        "quantity":     event_dict.get("quantity"),
+        "amount":       event_dict.get("amount"),
+        "payment_type": None,
+        "status":       event_dict.get("status", "open"),
+    }
+    append_row("Ledger", record)
+    logger.info("Excel sync → Ledger (credit): customer=%s", record.get("customer"))
+
+
+def append_inventory(record: dict) -> None:
+    """
+    Append a stock movement record to the Inventory sheet.
+
+    Args:
+        record: Dict with keys: timestamp, item, change, direction,
+                                reference_id, note
+    """
+    append_row("Inventory", record)
+    logger.info(
+        "Excel sync → Inventory: %s %s (%s)",
+        record.get("direction"), record.get("item"), record.get("change"),
+    )
+
+
+def append_invoice(record: dict) -> None:
+    """
+    Append an invoice record to the Invoices sheet.
+
+    Args:
+        record: Invoice dict from invoice_service.generate_invoice().
+    """
+    append_row("Invoices", record)
+    logger.info("Excel sync → Invoices: %s", record.get("invoice_id"))
+
+
+# ---------------------------------------------------------------------------
+# Convenience router — auto-dispatches based on orchestrator result intent
+# ---------------------------------------------------------------------------
+
+def sync_from_event(result: dict) -> None:
+    """
+    Automatically sync an orchestrator result to the correct Excel sheet(s).
+
+    Called by the FastAPI notification handler after skill execution so
+    the ledger is always up to date without skills needing to know about
+    the sync layer.
+
+    Note: Skills already write to Excel internally (Stages 5-6).
+    This function provides a secondary sync point for the FastAPI path
+    when skills are called via run_notiflow() in demo mode (where skills
+    don't execute and nothing is written). In live mode, this is a no-op
+    safety net — duplicate rows are avoided by checking event type.
+
+    Args:
+        result: Full orchestrator result dict
+                {message, intent, data, event}
+    """
+    intent = result.get("intent", "other")
+    event  = result.get("event", {})
+    event_name = event.get("event", "")
+
+    # In live mode the skill already wrote to Excel — skip to avoid duplicates.
+    # We only sync here if the result came from demo mode (event has no IDs).
+    order_data   = event.get("order")
+    payment_data = event.get("payment")
+    return_data  = event.get("return")
+    credit_data  = event.get("credit")
+
+    if intent == "order" and order_data and order_data.get("order_id"):
+        # Already written by skill — skip
+        logger.debug("sync_from_event: order already persisted by skill, skipping.")
+    elif intent == "payment" and payment_data and payment_data.get("entry_id"):
+        logger.debug("sync_from_event: payment already persisted by skill, skipping.")
+    elif intent == "return" and return_data and return_data.get("return_id"):
+        logger.debug("sync_from_event: return already persisted by skill, skipping.")
+    elif intent == "credit" and credit_data and credit_data.get("entry_id"):
+        logger.debug("sync_from_event: credit already persisted by skill, skipping.")
+    else:
+        logger.debug(
+            "sync_from_event: demo mode result or missing IDs — "
+            "no additional Excel write needed."
+        )

services/notification_generator.py

@@ -0,0 +1,109 @@
+"""
+services/notification_generator.py
+-----------------------------------
+Gemini-powered business notification generator for Notiflow.
+
+Purpose: generate realistic Hinglish business notifications for demo
+automation. This is entirely optional — the frontend simulation continues
+to work independently.
+
+Two modes:
+  1. Live Gemini generation  — calls Gemini API (requires GEMINI_API_KEY)
+  2. Static fallback pool    — returns from a hardcoded set when Gemini
+                               is unavailable (safe for offline demos)
+
+Public API
+----------
+get_notifications(n: int = 5) -> list[dict]
+    Returns a list of notification dicts:
+    [{"source": "whatsapp", "message": "..."}, ...]
+
+stream_notifications(n, delay_seconds) -> AsyncGenerator
+    Async generator yielding one notification at a time with a delay.
+    Used by the WebSocket endpoint.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import random
+from typing import AsyncGenerator
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Static fallback pool (used when Gemini is unavailable)
+# ---------------------------------------------------------------------------
+
+_FALLBACK_NOTIFICATIONS: list[dict] = [
+    {"source": "whatsapp", "message": "bhaiya 3 kurti bhej dena"},
+    {"source": "payment",  "message": "rahul ne 15000 bheja UPI se"},
+    {"source": "whatsapp", "message": "size chota hai exchange karna hai"},
+    {"source": "whatsapp", "message": "udhar me de dijiye"},
+    {"source": "amazon",   "message": "priya ke liye 2 kilo aata bhej dena"},
+    {"source": "payment",  "message": "amit bhai ka 8000 gpay se aaya"},
+    {"source": "whatsapp", "message": "3 kurti ka set ready rakhna"},
+    {"source": "return",   "message": "maal kharab tha wapas bhej diya"},
+    {"source": "whatsapp", "message": "suresh ko 500 ka maal udhar dena"},
+    {"source": "amazon",   "message": "order cancel karna hai, size bada hai"},
+    {"source": "payment",  "message": "50 piece pack karke rakhna kal tak"},
+    {"source": "whatsapp", "message": "geeta ke liye 5 metre kapda bhej dena"},
+]
+
+
+def _get_fallback(n: int) -> list[dict]:
+    """Return n randomly sampled notifications from the static pool."""
+    pool = _FALLBACK_NOTIFICATIONS * (n // len(_FALLBACK_NOTIFICATIONS) + 1)
+    return random.sample(pool, min(n, len(pool)))
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def get_notifications(n: int = 5) -> list[dict]:
+    """
+    Get n business notifications.
+
+    Tries Gemini first; falls back to static pool silently if unavailable.
+
+    Args:
+        n: Number of notifications to generate/return.
+
+    Returns:
+        List of {"source": str, "message": str} dicts.
+    """
+    try:
+        from models.gemini_client import generate_notifications
+        results = generate_notifications(n)
+        if results:
+            return results
+        logger.info("Gemini returned empty list — using fallback pool.")
+    except Exception as exc:
+        logger.info("Gemini unavailable (%s) — using fallback pool.", exc)
+
+    return _get_fallback(n)
+
+
+async def stream_notifications(
+    n: int = 5,
+    delay_seconds: float = 2.0,
+) -> AsyncGenerator[dict, None]:
+    """
+    Async generator that yields one notification at a time.
+
+    Fetches a fresh batch from get_notifications() then yields them
+    one-by-one with a configurable delay between each.
+
+    Args:
+        n:             Number of notifications per batch.
+        delay_seconds: Pause between yielded notifications.
+
+    Yields:
+        {"source": str, "message": str}
+    """
+    notifications = get_notifications(n)
+    for notification in notifications:
+        yield notification
+        await asyncio.sleep(delay_seconds)