bina soke kam kia

agent/extraction_agent.py

@@ -1,9 +1,136 @@
 """
-Extraction agent for extracting information from messages.
+Field extraction agent for Notiflow.
 """
 
-class ExtractionAgent:
-    """Agent for information extraction."""
-    
-    def __init__(self):
-        pass
+from __future__ import annotations
+
+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"],
+    "preparation": ["intent", "item", "quantity"],
+    "other": ["intent", "note"],
+}
+
+VALID_INTENTS = set(INTENT_SCHEMA.keys())
+logger = logging.getLogger(__name__)
+
+
+def _get_bedrock_client():
+    """Return a cached Bedrock runtime client."""
+    return get_bedrock_client()
+
+
+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,
+        },
+    }
+
+    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
+
+
+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:
+        parsed = json.loads(cleaned)
+    except json.JSONDecodeError:
+        match = re.search(r"\{.*\}", cleaned, re.DOTALL)
+        if match:
+            try:
+                parsed = json.loads(match.group(0))
+            except json.JSONDecodeError:
+                logger.warning("Could not parse Nova 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["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
+
+    return result
+
+
+def extract_fields(message: str, intent: str) -> dict:
+    """Extract structured business fields from a Hinglish message."""
+    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))}"
+        )
+
+    logger.info("Extracting fields | 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)
+    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"
+    schema_fields = INTENT_SCHEMA.get(intent, INTENT_SCHEMA["other"])
+    result = {field: None for field in schema_fields}
+    result["intent"] = intent
+    return result

agent/intent_agent.py

@@ -1,9 +1,90 @@
 """
-Intent agent for processing user intents.
+Intent detection agent for Notiflow.
 """
 
-class IntentAgent:
-    """Agent for intent detection and processing."""
-    
-    def __init__(self):
-        pass
+from __future__ import annotations
+
+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()
+
+
+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,
+        },
+    }
+
+    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()
+
+    try:
+        result = json.loads(cleaned)
+        intent = result.get("intent", "other").lower().strip()
+    except json.JSONDecodeError:
+        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)
+        intent = "other"
+
+    return {"intent": intent}
+
+
+def detect_intent(message: str) -> dict[str, str]:
+    """Detect the business intent of a Hinglish message."""
+    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("Detected intent: %s", result["intent"])
+    return result

agent/orchestrator.py

@@ -1,9 +1,117 @@
 """
-Orchestrator for managing agent workflows.
+orchestrator.py
+---------------
+Agent Orchestrator for Notiflow  (Stages 4 + 5 + FIX 2)
+
+Pipeline:
+    raw message
+        │
+        ▼
+    Intent Agent          →  detect intent
+        │
+        ▼
+    Extraction Agent      →  extract structured fields
+        │
+        ▼
+    Validator             →  normalise numbers, text, payment aliases  ← NEW
+        │
+        ▼
+    Skill Router          →  dispatch to business skill + persist
+        │
+        ▼
+    Structured Result     →  returned to caller
+
+Return shape (unchanged contract):
+    {
+        "message":  str,
+        "intent":   str,
+        "data":     dict,   — validated extracted fields
+        "event":    dict,   — skill event output
+    }
 """
 
-class Orchestrator:
-    """Orchestrates agent operations."""
-    
-    def __init__(self):
-        pass
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agent.intent_agent      import detect_intent
+from agent.extraction_agent  import extract_fields
+from validators.data_validator import validate_data
+from agent.router            import route_to_skill
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _build_result(
+    message:     str,
+    intent:      str,
+    validated:   dict,
+    skill_event: dict,
+) -> dict:
+    """Assemble the final result object returned to callers."""
+    return {
+        "message": message,
+        "intent":  intent,
+        "data":    validated,
+        "event":   skill_event,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def process_message(message: str) -> dict[str, Any]:
+    """
+    Run a raw business message through the full Notiflow agent pipeline.
+
+    Steps:
+        1. detect_intent   — Nova classifies the business intent
+        2. extract_fields  — Nova extracts structured entities
+        3. validate_data   — normalise numbers, text, payment aliases
+        4. route_to_skill  — dispatch to correct business skill + persist
+
+    Args:
+        message: Raw Hinglish or English business message.
+
+    Returns:
+        {
+            "message": str,
+            "intent":  str,
+            "data":    dict,
+            "event":   dict,
+        }
+
+    Raises:
+        ValueError:   Empty message.
+        RuntimeError: Agent or skill failure.
+    """
+    if not message or not message.strip():
+        raise ValueError("Message cannot be empty.")
+
+    message = message.strip()
+    logger.info("Orchestrator ← %r", message)
+
+    # ── Step 1: Intent ───────────────────────────────────────────────────────
+    intent = detect_intent(message)["intent"]
+    logger.info("Intent: %s", intent)
+
+    # ── Step 2: Extraction ───────────────────────────────────────────────────
+    extracted = extract_fields(message, intent)
+    raw_data  = {k: v for k, v in extracted.items() if k != "intent"}
+    logger.info("Extracted: %s", raw_data)
+
+    # ── Step 3: Validation ───────────────────────────────────────────────────
+    validated = validate_data(intent, raw_data)
+    logger.info("Validated: %s", validated)
+
+    # ── Step 4: Skill routing ────────────────────────────────────────────────
+    skill_event = route_to_skill(intent, validated)
+    logger.info("Skill event: %s", skill_event)
+
+    return _build_result(message, intent, validated, skill_event)

agent/router.py

@@ -1,9 +1,83 @@
 """
-Router for directing requests to appropriate handlers.
+router.py
+---------
+Stage 5: Skill Router for Notiflow
+
+The Skill Router is the decision layer between the Extraction Agent
+and the Business Skills. It receives a structured event (intent + data)
+and dispatches to the correct skill.
+
+Routing table:
+
+    intent          → skill function
+    ─────────────────────────────────
+    order           → process_order()
+    payment         → process_payment()
+    credit          → process_credit()
+    return          → process_return()
+    preparation     → process_preparation()
+    other           → (no skill; passthrough)
+
+If an intent has no registered skill the router returns a lightweight
+passthrough event so the pipeline never raises on unknown intents.
 """
 
-class Router:
-    """Routes requests to appropriate skills or services."""
-    
-    def __init__(self):
-        pass
+import logging
+from typing import Any
+
+from skills.order_skill       import process_order
+from skills.payment_skill     import process_payment
+from skills.credit_skill      import process_credit
+from skills.return_skill      import process_return
+from skills.preparation_skill import process_preparation
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Routing table  (intent → skill callable)
+# ---------------------------------------------------------------------------
+
+_SKILL_MAP: dict[str, Any] = {
+    "order":       process_order,
+    "payment":     process_payment,
+    "credit":      process_credit,
+    "return":      process_return,
+    "preparation": process_preparation,
+}
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def route_to_skill(intent: str, data: dict) -> dict:
+    """
+    Route a structured business event to the appropriate skill.
+
+    Args:
+        intent: The detected intent string (e.g. "payment", "order").
+        data:   The extracted field dict returned by the Extraction Agent
+                (without the "intent" key — that lives at the top level).
+
+    Returns:
+        A skill event dict.  Structure varies per skill but always contains
+        at minimum an "event" key describing what happened.
+
+        For unrecognised / "other" intents a passthrough dict is returned:
+        {"event": "unhandled", "intent": intent, "data": data}
+
+    Example:
+        >>> route_to_skill("payment", {"customer": "Rahul", "amount": 15000})
+        {
+            "event": "payment_recorded",
+            "payment": {"customer": "Rahul", "amount": 15000, "status": "received"}
+        }
+    """
+    skill_fn = _SKILL_MAP.get(intent)
+
+    if skill_fn is None:
+        logger.info("No skill registered for intent '%s' — returning passthrough.", intent)
+        return {"event": "unhandled", "intent": intent, "data": data}
+
+    logger.info("Routing intent '%s' to skill: %s", intent, skill_fn.__name__)
+    return skill_fn(data)

agent/skill_generator.py

@@ -0,0 +1,207 @@
+"""
+skill_generator.py
+------------------
+Dynamic Skill Generator for Notiflow.
+
+Creates new business skill Python files on demand and registers them in
+skills/skill_registry.json.
+
+Public API
+----------
+generate_skill(skill_name: str, description: str) -> dict
+list_skills() -> dict
+
+Safety rules:
+  - Raises SkillAlreadyExistsError if a skill with the same name exists.
+  - Skill names are normalised to snake_case.
+  - Generated files follow the standard skill template.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+from pathlib import Path
+from typing import Optional
+
+from app.config import ROOT, REGISTRY_FILE
+
+logger = logging.getLogger(__name__)
+
+SKILLS_DIR = ROOT / "skills"
+
+
+# ---------------------------------------------------------------------------
+# Exceptions
+# ---------------------------------------------------------------------------
+
+class SkillAlreadyExistsError(Exception):
+    """Raised when a skill with the given name already exists."""
+
+
+# ---------------------------------------------------------------------------
+# Skill file template
+# ---------------------------------------------------------------------------
+
+_SKILL_TEMPLATE = '''\
+"""
+{skill_name}.py
+{underline}
+Auto-generated business skill for Notiflow.
+
+Description: {description}
+
+Modify this file to implement the skill logic.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+logger = logging.getLogger(__name__)
+
+
+def {func_name}(data: dict) -> dict:
+    """
+    Execute the {display_name} skill.
+
+    Args:
+        data: Validated extraction dict from the orchestrator.
+
+    Returns:
+        Structured skill event dict.
+    """
+    logger.info("{display_name} skill executing: %s", data)
+
+    return {{
+        "event": "{event_name}",
+        "data":  data,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    }}
+'''
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _to_snake_case(name: str) -> str:
+    """Normalise skill name to snake_case (alphanumeric + underscores only)."""
+    name = name.strip().lower()
+    name = re.sub(r"[^a-z0-9]+", "_", name)
+    name = re.sub(r"_+", "_", name).strip("_")
+    return name
+
+
+def _load_registry() -> dict:
+    path = Path(REGISTRY_FILE)
+    if not path.exists():
+        return {}
+    try:
+        with path.open("r", encoding="utf-8") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, OSError) as exc:
+        logger.warning("Could not read registry: %s", exc)
+        return {}
+
+
+def _save_registry(registry: dict) -> None:
+    path = Path(REGISTRY_FILE)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8") as f:
+        json.dump(registry, f, indent=2, ensure_ascii=False)
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def generate_skill(skill_name: str, description: str) -> dict:
+    """
+    Generate a new skill file and register it.
+
+    Args:
+        skill_name:  Human-readable name (e.g. "discount_skill" or "Discount Skill").
+                     Normalised to snake_case automatically.
+        description: One-line description stored in the registry.
+
+    Returns:
+        Registry entry dict for the new skill:
+        {
+            "description": str,
+            "intent":      None,
+            "file":        "skills/<name>.py",
+            "builtin":     false
+        }
+
+    Raises:
+        SkillAlreadyExistsError: If a skill with the same name already exists
+                                 (either as a .py file or registry entry).
+        ValueError: If skill_name is empty or invalid.
+
+    Example:
+        >>> generate_skill("discount_skill", "Apply discount to an order")
+        {"description": "Apply discount...", "file": "skills/discount_skill.py", ...}
+    """
+    norm_name = _to_snake_case(skill_name)
+    if not norm_name:
+        raise ValueError(f"Invalid skill name: {skill_name!r}")
+
+    skill_file = SKILLS_DIR / f"{norm_name}.py"
+    registry   = _load_registry()
+
+    # ── Collision guard ──────────────────────────────────────────────────────
+    if norm_name in registry:
+        raise SkillAlreadyExistsError(
+            f"Skill '{norm_name}' already exists in the registry. "
+            "Choose a different name or delete the existing entry first."
+        )
+    if skill_file.exists():
+        raise SkillAlreadyExistsError(
+            f"Skill file '{skill_file}' already exists on disk. "
+            "Choose a different name or delete the existing file first."
+        )
+
+    # ── Generate file ───────���────────────────────────────────────────────────
+    display_name = norm_name.replace("_", " ").title()
+    func_name    = norm_name
+    event_name   = f"{norm_name}_executed"
+    underline    = "-" * (len(norm_name) + 3)     # matches "name.py" length
+
+    source = _SKILL_TEMPLATE.format(
+        skill_name   = norm_name,
+        underline    = underline,
+        description  = description,
+        func_name    = func_name,
+        display_name = display_name,
+        event_name   = event_name,
+    )
+
+    SKILLS_DIR.mkdir(parents=True, exist_ok=True)
+    skill_file.write_text(source, encoding="utf-8")
+    logger.info("Skill file created: %s", skill_file)
+
+    # ── Register ─────────────────────────────────────────────────────────────
+    entry = {
+        "description": description,
+        "intent":      None,          # caller can update after creation
+        "file":        f"skills/{norm_name}.py",
+        "builtin":     False,
+    }
+    registry[norm_name] = entry
+    _save_registry(registry)
+    logger.info("Skill '%s' registered.", norm_name)
+
+    return entry
+
+
+def list_skills() -> dict:
+    """
+    Return the full skill registry.
+
+    Returns:
+        Dict mapping skill_name → registry entry.
+    """
+    return _load_registry()

app/bedrock_client.py

@@ -0,0 +1,95 @@
+"""
+bedrock_client.py
+-----------------
+Reusable Amazon Bedrock runtime client for Notiflow.
+
+Both the Intent Agent and Extraction Agent import `call_nova()` from
+here instead of managing their own boto3 sessions.  The client is
+created once (lazy singleton) and reused across calls.
+
+Public API
+----------
+call_nova(prompt: str, max_tokens: int = 256) -> str
+    Send a plain-text prompt to Nova 2 Lite and return the response text.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from app.config import BEDROCK_MODEL_ID, BEDROCK_REGION
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Lazy singleton
+# ---------------------------------------------------------------------------
+
+_client = None
+
+
+def _get_client():
+    global _client
+    if _client is None:
+        try:
+            import boto3
+            _client = boto3.client(
+                service_name="bedrock-runtime",
+                region_name=BEDROCK_REGION,
+            )
+            logger.info("Bedrock client initialised (region=%s)", BEDROCK_REGION)
+        except Exception as exc:
+            raise RuntimeError(
+                f"Failed to create Bedrock client: {exc}\n"
+                "Check that boto3 is installed and AWS credentials are configured."
+            ) from exc
+    return _client
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def call_nova(prompt: str, max_tokens: int = 256) -> str:
+    """
+    Send a prompt to Amazon Nova 2 Lite via the Bedrock Converse API.
+
+    Args:
+        prompt:     Fully rendered prompt string.
+        max_tokens: Maximum tokens to generate (default 256).
+
+    Returns:
+        Raw text response from the model.
+
+    Raises:
+        RuntimeError: If the API call fails.
+    """
+    client = _get_client()
+
+    try:
+        response = client.converse(
+            modelId=BEDROCK_MODEL_ID,
+            messages=[
+                {
+                    "role": "user",
+                    "content": [{"text": prompt}],
+                }
+            ],
+            inferenceConfig={
+                "maxTokens": max_tokens,
+                "temperature": 0.0,
+                "topP": 1.0,
+            },
+        )
+        output_message = response["output"]["message"]
+        parts = [
+            block["text"]
+            for block in output_message["content"]
+            if "text" in block
+        ]
+        return " ".join(parts).strip()
+
+    except Exception as exc:
+        logger.error("Bedrock call failed: %s", exc)
+        raise RuntimeError(f"Nova API error: {exc}") from exc

app/config.py

@@ -1,7 +1,47 @@
 """
-Configuration settings for Notiflow application.
+config.py
+---------
+Central configuration for Notiflow.
+
+All file paths, feature flags, and model settings live here.
+Every other module imports from this file — no hardcoded paths elsewhere.
 """
 
-class Config:
-    """Base configuration class."""
-    pass
+from pathlib import Path
+
+# ---------------------------------------------------------------------------
+# Project root
+# ---------------------------------------------------------------------------
+
+ROOT = Path(__file__).parent.parent   # notiflow/
+
+# ---------------------------------------------------------------------------
+# Data paths
+# ---------------------------------------------------------------------------
+
+DATA_DIR      = ROOT / "data"
+DATA_FILE     = DATA_DIR / "notiflow_data.xlsx"   # Excel business store
+MEMORY_FILE   = DATA_DIR / "agent_memory.json"    # Agent memory (recent context)
+REGISTRY_FILE = ROOT / "skills" / "skill_registry.json"  # Skill registry
+
+# ---------------------------------------------------------------------------
+# Feature flags
+# ---------------------------------------------------------------------------
+
+# 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"
+
+# ---------------------------------------------------------------------------
+# Amazon Bedrock settings
+# ---------------------------------------------------------------------------
+
+BEDROCK_REGION   = os.getenv("AWS_REGION", "us-east-1")
+BEDROCK_MODEL_ID = "amazon.nova-lite-v1:0"
+
+# ---------------------------------------------------------------------------
+# Ensure data directory exists at import time
+# ---------------------------------------------------------------------------
+
+DATA_DIR.mkdir(parents=True, exist_ok=True)

app/main.py

@@ -1,11 +1,204 @@
 """
-Main entry point for the Notiflow application.
+main.py
+-------
+Primary entry point for Notiflow.
+
+Exposes run_notiflow(message) for programmatic use (dashboard, API, tests)
+and supports CLI testing directly from the terminal.
+
+Public API
+----------
+run_notiflow(message: str) -> dict
+
+    Runs the full pipeline and returns:
+    {
+        "message": str,
+        "intent":  str,
+        "data":    dict,
+        "event":   dict,
+    }
+
+CLI usage
+---------
+    python app/main.py "rahul ne 15000 bheja"
+
+    Prints the result as formatted JSON to stdout.
+
+Demo mode
+---------
+    Controlled by DEMO_MODE in app/config.py or the environment variable
+    NOTIFLOW_DEMO_MODE=false (set to disable demo mode).
+
+    When DEMO_MODE is True, a local simulation is used so the app works
+    without AWS credentials. The dashboard's DEMO_MODE toggle maps to this.
 """
 
-def main():
-    """Run the Notiflow application."""
-    pass
+from __future__ import annotations
+
+import json
+import logging
+import sys
+from typing import Any
+
+from app.config import DEMO_MODE
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Demo pipeline (no AWS needed)
+# ---------------------------------------------------------------------------
+
+_DEMO_RESPONSES: dict[str, dict] = {
+    "rahul ne 15000 bheja": {
+        "intent": "payment",
+        "data":   {"customer": "Rahul", "amount": 15000, "payment_type": None},
+        "event":  {"event": "payment_recorded",
+                   "payment": {"customer": "Rahul", "amount": 15000,
+                               "payment_type": None, "status": "received"}},
+    },
+    "bhaiya 3 kurti bhej dena": {
+        "intent": "order",
+        "data":   {"customer": None, "item": "kurti", "quantity": 3},
+        "event":  {"event": "order_received",
+                   "order":  {"customer": None, "item": "kurti",
+                              "quantity": 3, "status": "pending"},
+                   "invoice": {"invoice_id": "INV-DEMO-0001", "total_amount": 0.0}},
+    },
+    "priya ke liye 2 kilo aata bhej dena": {
+        "intent": "order",
+        "data":   {"customer": "Priya", "item": "aata", "quantity": 2},
+        "event":  {"event": "order_received",
+                   "order":  {"customer": "Priya", "item": "aata",
+                              "quantity": 2, "status": "pending"},
+                   "invoice": {"invoice_id": "INV-DEMO-0002", "total_amount": 0.0}},
+    },
+    "size chota hai exchange karna hai": {
+        "intent": "return",
+        "data":   {"customer": None, "item": None, "reason": "size issue"},
+        "event":  {"event": "return_requested",
+                   "return": {"customer": None, "item": None,
+                              "reason": "size issue", "status": "pending_review"}},
+    },
+    "udhar me de dijiye": {
+        "intent": "credit",
+        "data":   {"customer": None, "item": None, "quantity": None, "amount": None},
+        "event":  {"event": "credit_recorded",
+                   "credit": {"customer": None, "amount": None, "status": "open"}},
+    },
+    "suresh ko 500 ka maal udhar dena": {
+        "intent": "credit",
+        "data":   {"customer": "Suresh", "item": "goods", "quantity": None, "amount": 500},
+        "event":  {"event": "credit_recorded",
+                   "credit": {"customer": "Suresh", "amount": 500, "status": "open"}},
+    },
+    "3 kurti ka set ready rakhna": {
+        "intent": "preparation",
+        "data":   {"item": "kurti", "quantity": 3},
+        "event":  {"event": "preparation_queued",
+                   "preparation": {"item": "kurti", "quantity": 3, "status": "queued"}},
+    },
+    "amit bhai ka 8000 gpay se aaya": {
+        "intent": "payment",
+        "data":   {"customer": "Amit", "amount": 8000, "payment_type": "upi"},
+        "event":  {"event": "payment_recorded",
+                   "payment": {"customer": "Amit", "amount": 8000,
+                               "payment_type": "upi", "status": "received"}},
+    },
+}
+
+
+def _fallback_intent(message: str) -> str:
+    m = message.lower()
+    if any(w in m for w in ["bheja", "aaya", "cash", "gpay", "upi", "paytm", "online"]):
+        return "payment"
+    if any(w in m for w in ["exchange", "wapas", "return", "vapas", "size"]):
+        return "return"
+    if any(w in m for w in ["udhar", "credit", "baad"]):
+        return "credit"
+    if any(w in m for w in ["ready", "pack", "rakhna", "taiyar"]):
+        return "preparation"
+    if any(w in m for w in ["bhej", "dena", "chahiye", "kilo", "piece"]):
+        return "order"
+    return "other"
+
 
+def _run_demo(message: str) -> dict[str, Any]:
+    key      = message.strip().lower()
+    response = _DEMO_RESPONSES.get(key)
+    if response is None:
+        intent   = _fallback_intent(message)
+        response = {
+            "intent": intent,
+            "data":   {"note": f"Demo: classified as '{intent}'"},
+            "event":  {"event": f"{intent}_recorded",
+                       "note":  "Demo fallback — no exact match"},
+        }
+    return {
+        "message": message,
+        "intent":  response["intent"],
+        "data":    response["data"],
+        "event":   response["event"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def run_notiflow(message: str, demo_mode: bool | None = None) -> dict[str, Any]:
+    """
+    Run a business message through the full Notiflow pipeline.
+
+    This is the single function the dashboard and any external caller
+    should use. It never calls agents, skills, or Excel directly.
+
+    Args:
+        message:   Raw Hinglish or English business message.
+        demo_mode: Override DEMO_MODE from config. If None, uses config value.
+
+    Returns:
+        {
+            "message": str,
+            "intent":  str,
+            "data":    dict,
+            "event":   dict,
+        }
+
+    Raises:
+        ValueError:   Empty message.
+        RuntimeError: Pipeline failure (live mode only).
+    """
+    if not message or not message.strip():
+        raise ValueError("Message cannot be empty.")
+
+    use_demo = DEMO_MODE if demo_mode is None else demo_mode
+
+    if use_demo:
+        logger.info("run_notiflow [demo] ← %r", message)
+        return _run_demo(message.strip())
+    else:
+        logger.info("run_notiflow [live] ← %r", message)
+        from agent.orchestrator import process_message
+        return process_message(message.strip())
+
+
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
 
 if __name__ == "__main__":
-    main()
+    logging.basicConfig(level=logging.WARNING)
+
+    if len(sys.argv) < 2:
+        print("Usage: python app/main.py \"<business message>\"")
+        print('Example: python app/main.py "rahul ne 15000 bheja"')
+        sys.exit(1)
+
+    input_message = " ".join(sys.argv[1:])
+
+    try:
+        result = run_notiflow(input_message)
+        print(json.dumps(result, indent=2, ensure_ascii=False))
+    except Exception as exc:
+        print(json.dumps({"error": str(exc)}, indent=2))
+        sys.exit(1)

dashboard/streamlit_app.py

@@ -1,14 +1,335 @@
 """
-Streamlit dashboard application.
+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():
-    """Run the Streamlit dashboard."""
-    st.title("Notiflow Dashboard")
+    # 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()
+    main()

memory/agent_memory.py

@@ -0,0 +1,145 @@
+"""
+agent_memory.py
+---------------
+Agent memory layer for Notiflow.
+
+Stores recent business context so skills and future agents can reference
+what was last discussed (customer names, items, etc.).
+
+Storage: JSON file at the path defined in app/config.py (MEMORY_FILE).
+Structure:
+    {
+        "recent_customers": ["Rahul", "Priya"],   # newest last
+        "recent_items":     ["kurti", "aata"]
+    }
+
+Public API
+----------
+load_memory()  -> dict
+update_memory(customer=None, item=None) -> None
+
+Design notes:
+  - Maximum 10 entries per list (oldest pruned automatically).
+  - Read-modify-write is done in one function call to minimise race window.
+  - None values are silently ignored (no-op).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Optional
+
+from app.config import MEMORY_FILE
+
+logger = logging.getLogger(__name__)
+
+_MAX_ENTRIES = 10
+
+_EMPTY_MEMORY: dict = {
+    "recent_customers": [],
+    "recent_items":     [],
+}
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _read_file() -> dict:
+    """Read memory from disk; return empty structure if file missing/corrupt."""
+    path = Path(MEMORY_FILE)
+    if not path.exists():
+        return {k: list(v) for k, v in _EMPTY_MEMORY.items()}
+    try:
+        with path.open("r", encoding="utf-8") as f:
+            data = json.load(f)
+        # Ensure both keys are present even if file is partial
+        data.setdefault("recent_customers", [])
+        data.setdefault("recent_items", [])
+        return data
+    except (json.JSONDecodeError, OSError) as exc:
+        logger.warning("Could not read memory file (%s) — using empty memory.", exc)
+        return {k: list(v) for k, v in _EMPTY_MEMORY.items()}
+
+
+def _write_file(memory: dict) -> None:
+    """Write memory dict to disk atomically (write to temp then rename)."""
+    path    = Path(MEMORY_FILE)
+    tmp     = path.with_suffix(".tmp")
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with tmp.open("w", encoding="utf-8") as f:
+            json.dump(memory, f, indent=2, ensure_ascii=False)
+        tmp.replace(path)
+    except OSError as exc:
+        logger.error("Could not write memory file: %s", exc)
+        if tmp.exists():
+            tmp.unlink(missing_ok=True)
+
+
+def _append_unique(lst: list, value: str, max_size: int = _MAX_ENTRIES) -> list:
+    """
+    Append value to list, deduplicate, and keep only the most recent entries.
+    Most recent item is always at the end.
+    """
+    if value in lst:
+        lst.remove(value)          # remove old occurrence so it moves to end
+    lst.append(value)
+    return lst[-max_size:]         # keep newest max_size entries
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def load_memory() -> dict:
+    """
+    Load the current agent memory from disk.
+
+    Returns:
+        {
+            "recent_customers": [str, ...],
+            "recent_items":     [str, ...]
+        }
+    """
+    memory = _read_file()
+    logger.debug("Memory loaded: %s", memory)
+    return memory
+
+
+def update_memory(
+    customer: Optional[str] = None,
+    item:     Optional[str] = None,
+) -> None:
+    """
+    Update agent memory with a new customer name and/or item.
+
+    None values are silently ignored.
+    Duplicates are deduplicated and moved to the end (most recent position).
+
+    Args:
+        customer: Customer name to remember (e.g. "Rahul").
+        item:     Item name to remember (e.g. "kurti").
+
+    Example:
+        >>> update_memory(customer="Rahul", item="kurti")
+    """
+    if customer is None and item is None:
+        return
+
+    memory = _read_file()
+
+    if customer:
+        memory["recent_customers"] = _append_unique(
+            memory["recent_customers"], str(customer).strip()
+        )
+
+    if item:
+        memory["recent_items"] = _append_unique(
+            memory["recent_items"], str(item).strip()
+        )
+
+    _write_file(memory)
+    logger.info("Memory updated: customer=%s item=%s", customer, item)

parsers/message_parser.py

@@ -1,9 +1,26 @@
 """
-Message parser for processing incoming messages.
+Message normalization helpers for Notiflow.
 """
 
+from __future__ import annotations
+
+import re
+
+
 class MessageParser:
-    """Parses messages into structured formats."""
-    
-    def __init__(self):
-        pass
+    """Normalize incoming business messages before they reach the agents."""
+
+    _whitespace_pattern = re.compile(r"\s+")
+
+    def parse(self, message: str) -> str:
+        """Normalize a message for downstream LLM processing."""
+        if message is None:
+            return ""
+        normalized = str(message).strip().lower()
+        normalized = self._whitespace_pattern.sub(" ", normalized)
+        return normalized
+
+
+def parse_message(message: str) -> str:
+    """Convenience wrapper used by the backend entry point."""
+    return MessageParser().parse(message)

prompts/extraction_prompt.txt

@@ -1,9 +1,122 @@
-Information Extraction Prompt
+You are an AI data extraction agent for Notiflow, a business operations assistant for small businesses in India.
 
-Extract relevant entities and information from user messages.
+Your job is to extract structured business information from a message written in Hinglish (a mix of Hindi and English, often informal or colloquial).
 
-Instructions:
-- Identify key entities (names, dates, amounts, order numbers, etc.)
-- Extract relevant attributes for the detected intent
-- Structure the extracted data in JSON format
-- Mark confidence levels for extracted information
+You will be given:
+1. A business message
+2. The already-detected intent of the message
+
+## Your Task
+
+Extract all relevant business fields from the message based on the intent.
+Return ONLY a valid JSON object. No explanation. No markdown. No extra text.
+
+---
+
+## Supported Intents and Their Fields
+
+### order
+Extract:
+- customer (string | null): Name of the person placing the order, if mentioned
+- item (string | null): Product being ordered (translate/normalize to English)
+- quantity (number | null): How many units, kilos, sets, pieces, etc.
+
+### payment
+Extract:
+- customer (string | null): Name of the person who sent/received money
+- amount (number | null): The monetary amount (digits only, no currency symbol)
+- payment_type (string | null): Mode of payment if mentioned — "cash", "upi", "online", "cheque", or null
+
+### credit
+Extract:
+- customer (string | null): Name of the person taking goods on credit
+- item (string | null): Product being taken on credit, if mentioned
+- quantity (number | null): Quantity, if mentioned
+- amount (number | null): Credit amount if specified
+
+### return
+Extract:
+- customer (string | null): Name of the person returning the item, if mentioned
+- item (string | null): Product being returned or exchanged, if mentioned
+- reason (string | null): Reason for return — e.g. "size issue", "damaged", "wrong item"
+
+### preparation
+Extract:
+- item (string | null): Product to be prepared or packed
+- quantity (number | null): How many units to prepare
+
+### other
+Extract:
+- note (string | null): A short English summary of what the message says
+
+---
+
+## Hinglish Business Vocabulary Reference
+
+- "bhej dena" = send/deliver → order
+- "bheja" = sent (money) → payment
+- "ne bheja" = "[person] sent" → customer sent payment
+- "exchange karna" / "wapas karna" / "return karna" = return/exchange
+- "udhar" / "udhaar" = on credit
+- "ready rakhna" / "pack karna" = prepare/pack
+- "kilo", "kg" = kilogram quantity
+- "piece", "pcs", "nag" = unit quantity
+- "set" = a set/bundle
+- "kurti", "suit", "saree", "maal", "kapda" = clothing/fabric items
+- "chota" = small (size issue), "bada" = large
+- "number" can mean size (shoe/garment size)
+- "clear ho gaya" = payment cleared
+- "UPI", "paytm", "gpay", "phonepay", "online", "cash" = payment types
+
+---
+
+## Rules
+
+1. Always return a JSON object. Never return plain text.
+2. Always include the "intent" field using the intent you were given.
+3. If a field is not present in the message, set it to null.
+4. Never guess or hallucinate values not present in the message.
+5. Normalize names to Title Case (e.g., "rahul" → "Rahul").
+6. Normalize items to simple English nouns (e.g., "kurti" → "kurti", "maal" → "goods").
+7. Extract numbers as integers or floats, not strings.
+8. Do not include any field not listed for the given intent.
+
+---
+
+## Examples
+
+Intent: payment
+Message: "rahul ne 15000 bheja"
+Output: {"intent": "payment", "customer": "Rahul", "amount": 15000, "payment_type": null}
+
+Intent: order
+Message: "bhaiya 3 kurti bhej dena"
+Output: {"intent": "order", "customer": null, "item": "kurti", "quantity": 3}
+
+Intent: order
+Message: "priya ke liye 2 kilo aata bhej dena"
+Output: {"intent": "order", "customer": "Priya", "item": "aata", "quantity": 2}
+
+Intent: return
+Message: "size chota hai exchange karna hai"
+Output: {"intent": "return", "customer": null, "item": null, "reason": "size issue"}
+
+Intent: credit
+Message: "suresh ko udhar me 500 ka maal dena"
+Output: {"intent": "credit", "customer": "Suresh", "item": "goods", "quantity": null, "amount": 500}
+
+Intent: preparation
+Message: "3 kurti ka set ready rakhna"
+Output: {"intent": "preparation", "item": "kurti", "quantity": 3}
+
+Intent: payment
+Message: "amit bhai ka 8000 gpay se aaya"
+Output: {"intent": "payment", "customer": "Amit", "amount": 8000, "payment_type": "upi"}
+
+---
+
+Now extract fields from the following:
+
+Intent: {intent}
+Message: "{message}"
+Output:

prompts/intent_prompt.txt

@@ -1,8 +1,64 @@
-Intent Detection Prompt
+You are an AI agent for Notiflow, a business assistant for small businesses in India.
 
-Analyze the user message and determine the primary intent.
+Your job is to read a business message written in Hinglish (a mix of Hindi and English, often informal or colloquial) and classify its business intent.
 
-Instructions:
-- Identify the main action or request
-- Classify into categories: order, payment, credit, return, or other
-- Return the intent and confidence level
+## Supported Intents
+
+| Intent      | Description                                                                 |
+|-------------|-----------------------------------------------------------------------------|
+| order       | Customer wants to buy or order a product                                    |
+| payment     | Money has been received or sent by a customer or party                      |
+| credit      | Customer wants goods on credit (udhar), or an udhar transaction is recorded |
+| return      | Customer wants to exchange or return a product                              |
+| preparation | Shop owner needs to prepare, pack, or keep a set of items ready             |
+| other       | Message does not match any known business intent                            |
+
+## Hinglish Business Vocabulary Reference
+
+- "bhej dena" = send it / deliver it → likely an order
+- "bheja" = sent (money) → likely a payment
+- "exchange karna" / "wapas karna" = return/exchange → return
+- "udhar" / "udhaar" = on credit → credit
+- "ready rakhna" / "pack karna" = prepare/pack → preparation
+- "kilo", "piece", "set", "number" = quantity markers often found in orders
+- Names followed by amounts (e.g., "Rahul ne 500 bheja") → payment
+- "chota", "bada", "size" complaints → return
+
+## Rules
+
+1. Read the message carefully.
+2. Consider the full context, not just individual words.
+3. Always respond with a single JSON object and nothing else.
+4. Do not include any explanation, commentary, or markdown formatting.
+5. The JSON must have exactly one key: "intent".
+
+## Output Format
+
+{"intent": "<one of: order, payment, credit, return, preparation, other>"}
+
+## Examples
+
+Message: "bhaiya 2 kilo bhej dena"
+Output: {"intent": "order"}
+
+Message: "rahul ne 15000 bheja"
+Output: {"intent": "payment"}
+
+Message: "size chota hai exchange karna hai"
+Output: {"intent": "return"}
+
+Message: "udhar me de dijiye"
+Output: {"intent": "credit"}
+
+Message: "3 kurti ka set ready rakhna"
+Output: {"intent": "preparation"}
+
+Message: "aaj mausam bahut achha hai"
+Output: {"intent": "other"}
+
+---
+
+Now classify the following message:
+
+Message: "{message}"
+Output:

services/inventory_service.py

@@ -1,9 +1,146 @@
 """
-Inventory service for managing inventory.
+inventory_service.py
+--------------------
+Stage 6: Inventory Service for Notiflow
+
+Tracks stock movements as a delta log in the Inventory Excel sheet.
+Each event appends one row recording what changed, by how much,
+and in which direction (in / out).
+
+Design: delta-log (not current-stock snapshot)
+    - Every inventory change is a new row
+    - Current stock for an item = sum of all deltas for that item
+    - This keeps the history intact and avoids row-update complexity
+
+Directions:
+    "out"  — stock leaves (order fulfilled)
+    "in"   — stock arrives (return accepted, restock)
 """
 
-class InventoryService:
-    """Handles inventory-related operations."""
-    
-    def __init__(self):
-        pass
+import logging
+from datetime import datetime, timezone
+
+from utils.excel_writer import append_row, read_sheet
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def deduct_stock(item: str, quantity: int | float, reference_id: str, note: str = "") -> dict:
+    """
+    Record a stock deduction (items going out — e.g. an order is fulfilled).
+
+    Args:
+        item:         Name of the inventory item.
+        quantity:     Number of units being deducted.
+        reference_id: ID of the triggering record (e.g. order_id, invoice_id).
+        note:         Optional human-readable note.
+
+    Returns:
+        The inventory movement record that was persisted.
+
+    Example:
+        >>> deduct_stock("kurti", 3, "ORD-20240115-0001", "order fulfilled")
+        {
+            "timestamp": "...",
+            "item": "kurti",
+            "change": 3,
+            "direction": "out",
+            "reference_id": "ORD-20240115-0001",
+            "note": "order fulfilled"
+        }
+    """
+    if quantity is None or quantity <= 0:
+        logger.warning("deduct_stock called with invalid quantity: %s", quantity)
+        return {}
+
+    record = {
+        "timestamp":    _now_iso(),
+        "item":         item,
+        "change":       quantity,
+        "direction":    "out",
+        "reference_id": reference_id,
+        "note":         note or "stock deducted",
+    }
+
+    append_row("Inventory", record)
+    logger.info("Stock deducted: %s × %s (ref: %s)", quantity, item, reference_id)
+    return record
+
+
+def add_stock(item: str, quantity: int | float, reference_id: str, note: str = "") -> dict:
+    """
+    Record a stock addition (items coming in — e.g. a return is accepted).
+
+    Args:
+        item:         Name of the inventory item.
+        quantity:     Number of units being added.
+        reference_id: ID of the triggering record (e.g. return_id).
+        note:         Optional human-readable note.
+
+    Returns:
+        The inventory movement record that was persisted.
+    """
+    if quantity is None or quantity <= 0:
+        logger.warning("add_stock called with invalid quantity: %s", quantity)
+        return {}
+
+    record = {
+        "timestamp":    _now_iso(),
+        "item":         item,
+        "change":       quantity,
+        "direction":    "in",
+        "reference_id": reference_id,
+        "note":         note or "stock added",
+    }
+
+    append_row("Inventory", record)
+    logger.info("Stock added: %s × %s (ref: %s)", quantity, item, reference_id)
+    return record
+
+
+def get_stock_level(item: str) -> int | float:
+    """
+    Calculate the current stock level for an item by summing all deltas.
+
+    Args:
+        item: Name of the inventory item (case-insensitive match).
+
+    Returns:
+        Net stock level (int or float). Returns 0 if no records found.
+
+    Example:
+        >>> get_stock_level("kurti")
+        47
+    """
+    df = read_sheet("Inventory")
+
+    if df.empty or "item" not in df.columns:
+        return 0
+
+    item_rows = df[df["item"].str.lower() == item.lower()]
+
+    if item_rows.empty:
+        return 0
+
+    total = 0
+    for _, row in item_rows.iterrows():
+        change    = row.get("change", 0) or 0
+        direction = row.get("direction", "out")
+        if direction == "in":
+            total += change
+        else:
+            total -= change
+
+    return max(total, 0)   # Stock can't go below 0 in the display

services/invoice_service.py

@@ -1,9 +1,96 @@
 """
-Invoice service for managing invoices.
+invoice_service.py
+------------------
+Invoice generation service for Notiflow.
+
+Responsibility: generate a structured invoice object only.
+Excel persistence is handled separately by the skill layer.
+
+Invoice ID format: INV-YYYYMMDD-XXXX
+    - YYYYMMDD  today's UTC date
+    - XXXX      4-character alphanumeric suffix (uppercase)
+
+Public API
+----------
+generate_invoice(customer, item, quantity, unit_price=0.0) -> dict
 """
 
-class InvoiceService:
-    """Handles invoice-related operations."""
-    
-    def __init__(self):
-        pass
+from __future__ import annotations
+
+import random
+import string
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _make_invoice_id() -> str:
+    """
+    Format: INV-YYYYMMDD-XXXX
+    XXXX = random 4-char uppercase alphanumeric suffix.
+    No file read needed — random suffix avoids collisions for demo scale.
+    """
+    date_part = datetime.now(timezone.utc).strftime("%Y%m%d")
+    suffix    = "".join(random.choices(string.ascii_uppercase + string.digits, k=4))
+    return f"INV-{date_part}-{suffix}"
+
+
+def generate_invoice(
+    customer:   Optional[str],
+    item:       Optional[str],
+    quantity:   Optional[int | float],
+    unit_price: float = 0.0,
+    order_id:   Optional[str] = None,
+) -> dict:
+    """
+    Generate a structured invoice object.
+
+    Does NOT write to Excel — the calling skill persists the result.
+
+    Args:
+        customer:   Customer name (may be None).
+        item:       Item name (may be None).
+        quantity:   Quantity ordered (may be None).
+        unit_price: Price per unit. Defaults to 0.0.
+        order_id:   Optional linked order ID.
+
+    Returns:
+        {
+            "invoice_id":   "INV-20260315-AB12",
+            "timestamp":    ISO-8601 str,
+            "order_id":     str | None,
+            "customer":     str | None,
+            "item":         str | None,
+            "quantity":     int | float | None,
+            "unit_price":   float,
+            "total_amount": float,
+            "status":       "pending"
+        }
+    """
+    invoice_id   = _make_invoice_id()
+    qty          = quantity or 0
+    total_amount = round(float(qty) * unit_price, 2)
+
+    invoice = {
+        "invoice_id":   invoice_id,
+        "timestamp":    _now_iso(),
+        "order_id":     order_id,
+        "customer":     customer,
+        "item":         item,
+        "quantity":     quantity,
+        "unit_price":   unit_price,
+        "total_amount": total_amount,
+        "status":       "pending",
+    }
+
+    logger.info(
+        "Invoice generated: %s | customer=%s item=%s qty=%s total=%.2f",
+        invoice_id, customer, item, quantity, total_amount,
+    )
+    return invoice

skills/credit_skill.py

@@ -1,9 +1,69 @@
 """
-Credit management skill.
+credit_skill.py
+---------------
+Business Skill: Credit / Udhar  (Stage 6 — with persistence)
+
+Handles the "credit" intent.
+Appends a credit entry to the Ledger sheet.
+
+Expected input fields:
+    customer  (str | None)
+    item      (str | None)
+    quantity  (int | None)
+    amount    (int | None)
 """
 
-class CreditSkill:
-    """Handles credit-related operations."""
-    
-    def __init__(self):
-        pass
+import logging
+from datetime import datetime, timezone
+
+from utils.excel_writer import append_row, read_sheet
+
+logger = logging.getLogger(__name__)
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _generate_entry_id(prefix: str) -> str:
+    today = datetime.now(timezone.utc).strftime("%Y%m%d")
+    df    = read_sheet("Ledger")
+    seq   = len(df) + 1
+    return f"{prefix}-{today}-{seq:04d}"
+
+
+def process_credit(data: dict) -> dict:
+    """
+    Process a credit (udhar) event and append it to the Ledger sheet.
+
+    Args:
+        data: Extracted fields dict. Expected keys: customer, item, quantity, amount
+
+    Returns:
+        {
+            "event":  "credit_recorded",
+            "credit": { ledger entry }
+        }
+    """
+    logger.info("CreditSkill processing: %s", data)
+
+    entry_id = _generate_entry_id("CRD")
+
+    credit = {
+        "entry_id":     entry_id,
+        "timestamp":    _now_iso(),
+        "type":         "credit",
+        "customer":     data.get("customer"),
+        "item":         data.get("item"),
+        "quantity":     data.get("quantity"),
+        "amount":       data.get("amount"),
+        "payment_type": None,
+        "status":       "open",
+    }
+
+    append_row("Ledger", credit)
+
+    return {
+        "event":  "credit_recorded",
+        "credit": credit,
+    }

skills/order_skill.py

@@ -1,9 +1,96 @@
 """
-Order processing skill.
+order_skill.py
+--------------
+Business Skill: Order  (Stage 6 + UPGRADE 1 — memory update)
+
+On each order event this skill:
+    1. Appends an order record to the Orders sheet
+    2. Deducts stock from Inventory (delta log)
+    3. Generates an invoice object and saves it to Invoices sheet
+    4. Updates agent memory with customer + item        ← NEW (Upgrade 1)
+
+Expected input fields (all may be None if not captured):
+    customer  (str | None)
+    item      (str | None)
+    quantity  (int | None)
 """
 
-class OrderSkill:
-    """Handles order-related operations."""
-    
-    def __init__(self):
-        pass
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from utils.excel_writer         import append_row, read_sheet
+from services.invoice_service   import generate_invoice
+from services.inventory_service import deduct_stock
+from memory.agent_memory        import update_memory
+
+logger = logging.getLogger(__name__)
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _generate_order_id() -> str:
+    today = datetime.now(timezone.utc).strftime("%Y%m%d")
+    df    = read_sheet("Orders")
+    seq   = len(df) + 1
+    return f"ORD-{today}-{seq:04d}"
+
+
+def process_order(data: dict) -> dict:
+    """
+    Process an order event: persist order, update inventory, generate invoice,
+    and update agent memory.
+
+    Args:
+        data: Validated extraction dict. Keys: customer, item, quantity.
+
+    Returns:
+        {
+            "event":   "order_received",
+            "order":   { order record },
+            "invoice": { invoice record }
+        }
+    """
+    logger.info("OrderSkill ← %s", data)
+
+    customer = data.get("customer")
+    item     = data.get("item")
+    quantity = data.get("quantity")
+    order_id = _generate_order_id()
+
+    # 1 ── Persist order ──────────────────────────────────────────────────────
+    order = {
+        "order_id":  order_id,
+        "timestamp": _now_iso(),
+        "customer":  customer,
+        "item":      item,
+        "quantity":  quantity,
+        "status":    "pending",
+    }
+    append_row("Orders", order)
+
+    # 2 ── Inventory deduction ────────────────────────────────────────────────
+    if item and quantity:
+        deduct_stock(item, quantity, reference_id=order_id, note="order fulfilled")
+
+    # 3 ── Invoice generation ─────────────────────────────────────────────────
+    invoice = generate_invoice(
+        customer  = customer,
+        item      = item,
+        quantity  = quantity,
+        order_id  = order_id,
+        unit_price = 0.0,
+    )
+    append_row("Invoices", invoice)
+
+    # 4 ── Memory update ──────────────────────────────────────────────────────
+    update_memory(customer=customer, item=item)
+
+    return {
+        "event":   "order_received",
+        "order":   order,
+        "invoice": invoice,
+    }

skills/payment_skill.py

@@ -1,9 +1,68 @@
 """
-Payment processing skill.
+payment_skill.py
+----------------
+Business Skill: Payment  (Stage 6 — with persistence)
+
+Handles the "payment" intent.
+Appends a payment entry to the Ledger sheet.
+
+Expected input fields:
+    customer      (str | None)  — name of the person who sent money
+    amount        (int | None)  — monetary amount
+    payment_type  (str | None)  — "cash", "upi", "online", "cheque", or None
 """
 
-class PaymentSkill:
-    """Handles payment-related operations."""
-    
-    def __init__(self):
-        pass
+import logging
+from datetime import datetime, timezone
+
+from utils.excel_writer import append_row, read_sheet
+
+logger = logging.getLogger(__name__)
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _generate_entry_id(prefix: str) -> str:
+    today = datetime.now(timezone.utc).strftime("%Y%m%d")
+    df    = read_sheet("Ledger")
+    seq   = len(df) + 1
+    return f"{prefix}-{today}-{seq:04d}"
+
+
+def process_payment(data: dict) -> dict:
+    """
+    Process a payment event and append it to the Ledger sheet.
+
+    Args:
+        data: Extracted fields dict. Expected keys: customer, amount, payment_type
+
+    Returns:
+        {
+            "event":   "payment_recorded",
+            "payment": { ledger entry }
+        }
+    """
+    logger.info("PaymentSkill processing: %s", data)
+
+    entry_id = _generate_entry_id("PAY")
+
+    payment = {
+        "entry_id":     entry_id,
+        "timestamp":    _now_iso(),
+        "type":         "payment",
+        "customer":     data.get("customer"),
+        "item":         None,
+        "quantity":     None,
+        "amount":       data.get("amount"),
+        "payment_type": data.get("payment_type"),
+        "status":       "received",
+    }
+
+    append_row("Ledger", payment)
+
+    return {
+        "event":   "payment_recorded",
+        "payment": payment,
+    }

skills/preparation_skill.py

@@ -0,0 +1,117 @@
+"""
+preparation_skill.py
+--------------------
+Business Skill: Preparation / Inventory Pack  (Stage 6 — with persistence)
+
+Handles the "preparation" intent.
+Appends a preparation task to the Inventory sheet as a "reserved" movement.
+
+This records that stock is being set aside / packed, without fully
+deducting it (deduction happens when the linked order ships).
+If no linked order exists (standalone prep task), the record still logs
+the intention for the shop owner's reference.
+
+Expected input fields:
+    item      (str | None)  — item to prepare or pack
+    quantity  (int | None)  — number of units to prepare
+"""
+
+import logging
+from datetime import datetime, timezone
+
+from utils.excel_writer import append_row, read_sheet
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _generate_prep_id() -> str:
+    """Generate a sequential preparation ID: PREP-YYYYMMDD-XXXX."""
+    today = datetime.now(timezone.utc).strftime("%Y%m%d")
+    # Count existing preparation entries in Inventory to sequence the ID
+    df  = read_sheet("Inventory")
+    prep_rows = df[df["direction"] == "reserved"] if not df.empty and "direction" in df.columns else df
+    seq = len(prep_rows) + 1
+    return f"PREP-{today}-{seq:04d}"
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def process_preparation(data: dict) -> dict:
+    """
+    Process a preparation / packing task and log it to the Inventory sheet.
+
+    The movement is logged with direction="reserved" so it is visible in
+    the inventory log but does not reduce the available stock count until
+    the items actually ship.
+
+    Args:
+        data: Extracted fields dict from the Extraction Agent.
+              Expected keys: item, quantity
+
+    Returns:
+        {
+            "event":       "preparation_queued",
+            "preparation": {
+                "prep_id":   str,
+                "timestamp": ISO-8601 str,
+                "item":      str | None,
+                "quantity":  int | None,
+                "status":    "queued"
+            }
+        }
+
+    Example:
+        >>> process_preparation({"item": "kurti", "quantity": 3})
+        {
+            "event": "preparation_queued",
+            "preparation": {
+                "prep_id": "PREP-20240115-0001",
+                "timestamp": "...",
+                "item": "kurti",
+                "quantity": 3,
+                "status": "queued"
+            }
+        }
+    """
+    logger.info("PreparationSkill processing: %s", data)
+
+    item     = data.get("item")
+    quantity = data.get("quantity")
+    prep_id  = _generate_prep_id()
+
+    # Log to Inventory sheet as a "reserved" movement
+    if item:
+        inventory_record = {
+            "timestamp":    _now_iso(),
+            "item":         item,
+            "change":       quantity or 0,
+            "direction":    "reserved",
+            "reference_id": prep_id,
+            "note":         "preparation task queued",
+        }
+        append_row("Inventory", inventory_record)
+
+    prep = {
+        "prep_id":   prep_id,
+        "timestamp": _now_iso(),
+        "item":      item,
+        "quantity":  quantity,
+        "status":    "queued",
+    }
+
+    logger.info("Preparation task logged: %s", prep_id)
+
+    return {
+        "event":       "preparation_queued",
+        "preparation": prep,
+    }

skills/return_skill.py

@@ -1,9 +1,103 @@
 """
-Return processing skill.
+return_skill.py
+---------------
+Business Skill: Return / Exchange  (Stage 6 — with persistence)
+
+Handles the "return" intent.
+Appends a return record to the Returns sheet in notiflow_data.xlsx.
+
+Inventory is NOT updated here. Stock is only added back once the return
+status changes to "approved" — to be handled in a future stage via a
+status-update workflow.
+
+Expected input fields:
+    customer  (str | None)  — customer making the return
+    item      (str | None)  — item being returned or exchanged
+    reason    (str | None)  — reason for return (e.g. "size issue", "damaged")
 """
 
-class ReturnSkill:
-    """Handles return-related operations."""
-    
-    def __init__(self):
-        pass
+import logging
+from datetime import datetime, timezone
+
+from utils.excel_writer import append_row, read_sheet
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _generate_return_id() -> str:
+    """Generate a sequential return ID: RET-YYYYMMDD-XXXX."""
+    today = datetime.now(timezone.utc).strftime("%Y%m%d")
+    df    = read_sheet("Returns")
+    seq   = len(df) + 1
+    return f"RET-{today}-{seq:04d}"
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def process_return(data: dict) -> dict:
+    """
+    Process a return / exchange event and persist it to the Returns sheet.
+
+    Inventory is NOT updated here — stock is only restored after approval.
+
+    Args:
+        data: Extracted fields dict from the Extraction Agent.
+              Expected keys: customer, item, reason
+
+    Returns:
+        {
+            "event":  "return_requested",
+            "return": {
+                "return_id": str,
+                "timestamp": ISO-8601 str,
+                "customer":  str | None,
+                "item":      str | None,
+                "reason":    str | None,
+                "status":    "pending_review"
+            }
+        }
+
+    Example:
+        >>> process_return({"customer": None, "item": None, "reason": "size issue"})
+        {
+            "event": "return_requested",
+            "return": {
+                "return_id": "RET-20240115-0001",
+                "timestamp": "...",
+                "customer": None,
+                "item": None,
+                "reason": "size issue",
+                "status": "pending_review"
+            }
+        }
+    """
+    logger.info("ReturnSkill processing: %s", data)
+
+    return_id = _generate_return_id()
+
+    return_entry = {
+        "return_id": return_id,
+        "timestamp": _now_iso(),
+        "customer":  data.get("customer"),
+        "item":      data.get("item"),
+        "reason":    data.get("reason"),
+        "status":    "pending_review",
+    }
+
+    append_row("Returns", return_entry)
+    logger.info("Return logged: %s", return_id)
+
+    return {
+        "event":  "return_requested",
+        "return": return_entry,
+    }

utils/excel_writer.py

@@ -1,9 +1,164 @@
 """
-Excel file writing utilities.
+utils/excel_writer.py
+---------------------
+Excel persistence utility for Notiflow.  (FIX 4 — config-driven paths)
+
+All file paths come from app/config.py — no hardcoded paths in this module.
+
+Single-file, multi-sheet store:  DATA_FILE  (default: data/notiflow_data.xlsx)
+
+Sheets and their column schemas:
+    Orders     — order records
+    Ledger     — payment and credit entries
+    Returns    — return / exchange requests
+    Inventory  — stock movement delta log
+    Invoices   — generated invoice records
+
+Public API
+----------
+append_row(sheet_name, record)          — append one row, atomic save
+append_rows(sheet_name, records)        — append many rows, one save
+read_sheet(sheet_name) -> DataFrame     — read sheet into pandas DataFrame
 """
 
-class ExcelWriter:
-    """Writes data to Excel files."""
-    
-    def __init__(self):
-        pass
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+
+import pandas as pd
+from openpyxl import load_workbook, Workbook
+
+from app.config import DATA_FILE   # FIX 4: single source of truth
+
+logger = logging.getLogger(__name__)
+
+EXCEL_FILE = Path(DATA_FILE)        # re-export for modules that imported it directly
+
+# ---------------------------------------------------------------------------
+# Canonical column schemas
+# ---------------------------------------------------------------------------
+
+SHEET_SCHEMAS: dict[str, list[str]] = {
+    "Orders": [
+        "order_id", "timestamp", "customer", "item", "quantity", "status"
+    ],
+    "Ledger": [
+        "entry_id", "timestamp", "type", "customer", "item",
+        "quantity", "amount", "payment_type", "status"
+    ],
+    "Returns": [
+        "return_id", "timestamp", "customer", "item", "reason", "status"
+    ],
+    "Inventory": [
+        "timestamp", "item", "change", "direction", "reference_id", "note"
+    ],
+    "Invoices": [
+        "invoice_id", "timestamp", "order_id", "customer",
+        "item", "quantity", "unit_price", "total_amount", "status"
+    ],
+}
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _ensure_file() -> None:
+    """Create the Excel file with all sheets if it does not exist."""
+    if EXCEL_FILE.exists():
+        return
+
+    EXCEL_FILE.parent.mkdir(parents=True, exist_ok=True)
+    logger.info("Creating new Excel file: %s", EXCEL_FILE)
+    wb = Workbook()
+
+    if "Sheet" in wb.sheetnames:
+        del wb["Sheet"]
+
+    for sheet_name, columns in SHEET_SCHEMAS.items():
+        ws = wb.create_sheet(title=sheet_name)
+        ws.append(columns)
+
+    wb.save(EXCEL_FILE)
+    logger.info("Excel file created with sheets: %s", list(SHEET_SCHEMAS.keys()))
+
+
+def _ensure_sheet(wb: Workbook, sheet_name: str) -> None:
+    if sheet_name not in wb.sheetnames:
+        ws = wb.create_sheet(title=sheet_name)
+        columns = SHEET_SCHEMAS.get(sheet_name, [])
+        if columns:
+            ws.append(columns)
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def append_row(sheet_name: str, record: dict) -> None:
+    """
+    Append a single record to the named Excel sheet.
+
+    Missing columns default to None. Extra keys in record are ignored.
+    File and sheet are created automatically if they don't exist.
+
+    Args:
+        sheet_name: e.g. "Orders", "Ledger"
+        record:     Dict of column → value pairs.
+
+    Raises:
+        ValueError: Unknown sheet_name.
+    """
+    if sheet_name not in SHEET_SCHEMAS:
+        raise ValueError(
+            f"Unknown sheet '{sheet_name}'. Valid: {list(SHEET_SCHEMAS)}"
+        )
+
+    _ensure_file()
+    wb = load_workbook(EXCEL_FILE)
+    _ensure_sheet(wb, sheet_name)
+
+    ws      = wb[sheet_name]
+    columns = SHEET_SCHEMAS[sheet_name]
+    ws.append([record.get(col) for col in columns])
+
+    wb.save(EXCEL_FILE)
+    logger.debug("Row appended to '%s': %s", sheet_name, record)
+
+
+def append_rows(sheet_name: str, records: list[dict]) -> None:
+    """Append multiple records in one file open/save cycle."""
+    if not records:
+        return
+    if sheet_name not in SHEET_SCHEMAS:
+        raise ValueError(f"Unknown sheet '{sheet_name}'.")
+
+    _ensure_file()
+    wb = load_workbook(EXCEL_FILE)
+    _ensure_sheet(wb, sheet_name)
+
+    ws      = wb[sheet_name]
+    columns = SHEET_SCHEMAS[sheet_name]
+    for record in records:
+        ws.append([record.get(col) for col in columns])
+
+    wb.save(EXCEL_FILE)
+    logger.debug("Appended %d rows to '%s'", len(records), sheet_name)
+
+
+def read_sheet(sheet_name: str) -> pd.DataFrame:
+    """
+    Read a sheet into a DataFrame.
+
+    Returns an empty DataFrame (with correct columns) if the file or
+    sheet does not exist yet.
+    """
+    columns = SHEET_SCHEMAS.get(sheet_name, [])
+    if not EXCEL_FILE.exists():
+        return pd.DataFrame(columns=columns)
+    try:
+        return pd.read_excel(EXCEL_FILE, sheet_name=sheet_name)
+    except Exception as exc:
+        logger.warning("Could not read sheet '%s': %s", sheet_name, exc)
+        return pd.DataFrame(columns=columns)

validators/data_validator.py

@@ -1,9 +1,128 @@
 """
-Data validation utilities.
+data_validator.py
+-----------------
+Validation and normalization utilities for extracted business data.
+
+Used in the orchestrator pipeline between extraction and skill routing:
+
+    extract_fields() → validate_data() → route_to_skill()
+
+The validator normalises:
+  - text fields (customer, item, reason)  → stripped, lowercased or title-cased
+  - payment_type aliases (gpay → upi, paytm → upi, etc.)
+  - numeric fields (amount, quantity)     → int or float, never negative
 """
 
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+_NUMBER_PATTERN = re.compile(r"-?\d+(?:\.\d+)?")
+
+
 class DataValidator:
-    """Validates data against specified schemas."""
-    
-    def __init__(self):
-        pass
+    """Validate and normalize extraction-agent output."""
+
+    def validate(self, intent: str, data: dict[str, Any]) -> dict[str, Any]:
+        """Return a cleaned copy of extracted data for the given intent."""
+        cleaned = dict(data or {})
+
+        if "customer" in cleaned:
+            cleaned["customer"] = self._clean_text(cleaned.get("customer"), title=True)
+        if "item" in cleaned:
+            cleaned["item"] = self._clean_text(cleaned.get("item"))
+        if "reason" in cleaned:
+            cleaned["reason"] = self._clean_text(cleaned.get("reason"))
+        if "payment_type" in cleaned:
+            cleaned["payment_type"] = self._normalize_payment_type(cleaned.get("payment_type"))
+        if "amount" in cleaned:
+            cleaned["amount"] = self._to_number(cleaned.get("amount"), as_int_if_possible=True)
+        if "quantity" in cleaned:
+            cleaned["quantity"] = self._to_number(cleaned.get("quantity"), as_int_if_possible=True)
+
+        # Business rules: amounts and quantities must be positive
+        if intent == "payment" and cleaned.get("amount") is not None and cleaned["amount"] < 0:
+            cleaned["amount"] = abs(cleaned["amount"])
+        if (
+            intent in {"order", "credit", "preparation"}
+            and cleaned.get("quantity") is not None
+            and cleaned["quantity"] < 0
+        ):
+            cleaned["quantity"] = abs(cleaned["quantity"])
+
+        return cleaned
+
+    @staticmethod
+    def _clean_text(value: Any, *, title: bool = False) -> str | None:
+        if value is None:
+            return None
+        text = str(value).strip()
+        if not text:
+            return None
+        text = re.sub(r"\s+", " ", text)
+        return text.title() if title else text.lower()
+
+    @staticmethod
+    def _normalize_payment_type(value: Any) -> str | None:
+        text = DataValidator._clean_text(value)
+        if text is None:
+            return None
+
+        aliases = {
+            "gpay":          "upi",
+            "google pay":    "upi",
+            "phonepe":       "upi",
+            "phone pe":      "upi",
+            "paytm":         "upi",
+            "upi":           "upi",
+            "cash":          "cash",
+            "online":        "online",
+            "bank transfer": "online",
+            "neft":          "online",
+            "imps":          "online",
+            "rtgs":          "online",
+            "cheque":        "cheque",
+            "check":         "cheque",
+        }
+        return aliases.get(text, text)
+
+    @staticmethod
+    def _to_number(value: Any, *, as_int_if_possible: bool = False) -> int | float | None:
+        if value is None or value == "":
+            return None
+        if isinstance(value, (int, float)) and not isinstance(value, bool):
+            number = float(value)
+        else:
+            text  = str(value).replace(",", "").lower()
+            match = _NUMBER_PATTERN.search(text)
+            if not match:
+                return None
+            number = float(match.group(0))
+        if as_int_if_possible and number.is_integer():
+            return int(number)
+        return number
+
+
+# ---------------------------------------------------------------------------
+# Convenience wrapper — used by the orchestrator
+# ---------------------------------------------------------------------------
+
+def validate_data(intent: str, data: dict[str, Any]) -> dict[str, Any]:
+    """
+    Normalise and validate extracted data for the given intent.
+
+    This is the function the orchestrator imports:
+
+        from validators.data_validator import validate_data
+        cleaned = validate_data(intent, raw_data)
+
+    Args:
+        intent: Detected intent string (e.g. "payment", "order").
+        data:   Raw extraction dict from the Extraction Agent.
+
+    Returns:
+        Cleaned, normalised copy of the data dict.
+    """
+    return DataValidator().validate(intent, data)