Hi ${props.name},
-- Your purchase request ${props.prNumber} - for ₹${props.totalAmount.toLocaleString('en-IN')} - has been ${props.decision} - by ${props.approverName}. -
- ${props.comments - ? `Comment: "${props.comments}"
` - : '' - } - ${approved - ? `The procurement team will now raise a - Purchase Order with the vendor. Expect - delivery within the lead time stated - on each item.
` - : `Please sign in to ProcureAI to revise - your request or contact your manager for - more information.
` - } - - ` - }) - } - -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ -ENV CHECKLIST (everything needed) -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ - - # Upstash — from console.upstash.com - UPSTASH_REDIS_REST_URL= - UPSTASH_REDIS_REST_TOKEN= - - # QStash — from console.upstash.com → QStash - QSTASH_TOKEN= - QSTASH_CURRENT_SIGNING_KEY= - QSTASH_NEXT_SIGNING_KEY= - - # Your public URL (QStash webhook target) - NEXT_PUBLIC_APP_URL=https://procureai.yourdomain.com - - # Resend (email) - RESEND_API_KEY=re_xxx - - # Dev only — expose localhost to QStash - # ngrok http 3000 → copy https URL to NEXT_PUBLIC_APP_URL - -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ -GIT COMMITS -━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ - - git commit -m "feat(schema): add expoPushToken to User" - git commit -m "feat(notifications): Upstash pub/sub + QStash wiring" - git commit -m "feat(api): /notifications/subscribe SSE endpoint" - git commit -m "feat(api): /notifications/deliver QStash handler" - git commit -m "feat(api): /notifications/seen mark-read endpoint" - git commit -m "feat(web): usePRNotifications hook + toast" - git commit -m "feat(mobile): Expo push token registration" - git commit -m "feat(email): sendApprovalEmail via Resend" -``` - -*** - -## Why This Stack Combination Is Correct - -**Upstash Redis pub/sub** uses HTTP-streaming SSE under the hood — not a persistent TCP socket — which means it works in serverless Next.js, Vercel Edge, and Azure Container Apps without any infra changes. The `Subscriber` class handles reconnection automatically. [mintlify](https://www.mintlify.com/upstash/redis-js/advanced/pub-sub) - -**QStash is not a real-time tool** — it's a durable message queue with HTTP delivery and retries. It fills the gap pub/sub can't: if the employee's tab is closed, Redis PUBLISH is a no-op and the notification would be lost forever. QStash's 5-second delay + 3 retries guarantees it always reaches *something* (email or push). [upstash](https://upstash.com/docs/qstash/features/delay) - -**The `pr:seen:{prId}` key** is the coordination mechanism between the two systems. The Redis pub/sub path sets it; the QStash path checks it. This means you never send a duplicate notification — the employee never gets both a toast *and* an email for the same event. \ No newline at end of file + + +------ + +## PRD Overview + +**Product name:** SupportPilot + +**One-line description:** An AI-powered customer support workspace that helps support agents search Salesforce, retrieve customer context, draft replies, create and update cases, and escalate to humans with approval-style HITL checkpoints.[youtube](https://www.youtube.com/watch?v=v5iSo5fglV8)[help.salesforce](https://help.salesforce.com/s/articleView?id=release-notes.rn_asp_ga.htm&language=es&release=254&type=5) + +**Primary audience:** Support agents, team leads, and support ops teams at B2B SaaS companies that use Salesforce Service Cloud. Salesforce case management is explicitly built around collecting, tracking, assigning, and resolving customer issues in one system.speridian+1 + +**Portfolio goal:** Demonstrate applied AI, agentic workflows, legacy system integration, RAG over support knowledge, and production-quality GenUI.[help.salesforce](https://help.salesforce.com/s/articleView?id=release-notes.rn_asp_ga.htm&language=es&release=254&type=5)[youtube](https://www.youtube.com/watch?v=36tz6V_7Xpc) + +------ + +## Core Experience + +The product starts as a chat-style support cockpit. A user can type a customer issue, search Salesforce cases, inspect account history, find similar past tickets, draft a response, and create or update a case without leaving the interface. Salesforce support flows commonly include case creation, case tracking, severity changes, and transcript-backed summaries, which makes this interaction model feel native to the platform.[help.salesforce](https://help.salesforce.com/s/articleView?id=release-notes.rn_einstein_create_case_with_enhanced_data.htm&language=sv&release=256&type=5)[youtube](https://www.youtube.com/watch?v=v5iSo5fglV8) + +The agent should be able to answer questions like: + +- “Show me open cases for Acme.” +- “Summarize the last three cases from this customer.” +- “Draft a reply and create a case.” +- “Escalate this to tier 2.” +- “What is the status of case 00012345?”[youtube](https://www.youtube.com/watch?v=v5iSo5fglV8)[speridian](https://speridian.com/blogs/the-top-features-of-salesforce-case-management/) + +------ + +## Functional Scope + +## F1 — Authentication and session + +- Email/password login. +- Supabase session for the web app. +- Role-aware routing for support agent, team lead, support ops, and admin. +- Session context must include user id, role, queue/team, and Salesforce org mapping. + +## F2 — Support chat workspace + +- Main chat at `/support`. +- Conversational entry for all support actions. +- Persistent conversation history. +- Streaming assistant responses. +- Reply drafts shown as editable GenUI cards. + +## F3 — Salesforce case search + +- Search Salesforce cases by customer name, email, case number, subject, status, priority, and owner. +- Support natural language queries and structured filters. +- Return case lists in a `CaseListCard` GenUI component. + +## F4 — Salesforce customer context + +- Fetch account, contact, and case history. +- Show customer tier, open cases, last reply date, and recent interactions. +- Summarize customer context in a `CustomerContextCard`. + +## F5 — Similar ticket retrieval + +- Search past resolved cases and knowledge articles. +- Use RAG over internal support docs and case transcripts. +- Return concise suggestions with citations or supporting snippets. + +## F6 — Draft reply generation + +- Generate a suggested reply grounded in case data and KB context. +- Human can edit before sending. +- Draft appears in a `ReplyDraftCard`. + +## F7 — Case creation and update + +- Create cases from chat. +- Update subject, description, priority, status, owner, and comments. +- Support transcript attachment where available. +- Case creation flow should mirror Salesforce support patterns that ask for issue details and produce a structured case summary.[youtube](https://www.youtube.com/watch?v=OUxtejvgL7Y)[help.salesforce](https://help.salesforce.com/s/articleView?id=release-notes.rn_einstein_create_case_with_enhanced_data.htm&language=sv&release=256&type=5) + +## F8 — Escalation and HITL + +- Certain actions require confirmation before execution: + - closing a case, + - escalating priority, + - reassigning ownership, + - sending an external reply. +- Use interrupt/resume workflow for human approval. +- Show a clear approval card before mutation. + +## F9 — Team lead dashboard + +- Pending escalations. +- SLA risk view. +- Open cases by queue. +- Recent agent actions. +- Cases awaiting approval or manual review. + +## F10 — Support ops dashboard + +- Case volume trend. +- Resolution time. +- SLA compliance. +- Case categories. +- Escalation rate. +- Knowledge article effectiveness. + +## F11 — Notifications + +- Notify team leads on escalations. +- Notify agents on approvals or case updates. +- Notify ops on SLA breach risk. +- Display real-time notification bell in the UI. + +## F12 — Third-party integrations + +- Salesforce REST API for Cases, Accounts, Contacts, and Notes. +- Salesforce SOQL search for records. +- Knowledge base retrieval from internal documents. +- Optional Jira integration for engineering escalations. +- Optional email or Slack handoff for follow-ups. + +## F13 — Observability and audit + +- Log every tool call. +- Save agent traces. +- Capture approval decisions. +- Capture failed queries and fallback behavior. +- Use Langfuse or equivalent tracing. + +## F14 — Evaluation harness + +- Unit tests for UI states. +- Integration tests for Salesforce tool calls. +- E2E tests for login, search, draft reply, create case, and escalation. +- Eval set for ambiguous support scenarios. + +## F15 — Security and permissions + +- Support agents can search and draft. +- Team leads can approve escalations and case changes. +- Admin can manage mappings and integrations. +- Tool access is filtered by role. +- Sensitive customer data is never exposed outside authorized views. + +------ + +## Tooling Spec + +## Required agent tools + +- `search_salesforce_cases(query, filters)` +- `get_case_details(case_id)` +- `get_customer_context(account_or_contact_id)` +- `search_similar_tickets(query)` +- `search_knowledge_base(query)` +- `draft_case_reply(case_id, context)` +- `create_case(subject, description, priority, account_id)` +- `update_case(case_id, fields)` +- `escalate_case(case_id, reason)` +- `send_case_reply(case_id, message)` +- `link_jira_issue(case_id, summary)` optional + +Salesforce’s own docs and integrations support case search, case creation, and agent workflows, and LangChain has Salesforce integration docs available for tool building.docs.langchain+2 + +------ + +## GenUI Components + +- `CaseListCard` +- `CustomerContextCard` +- `ReplyDraftCard` +- `EscalationCard` +- `SlaGauge` +- `SupportOpsChart` +- `NotificationBell` + +Every component must handle loading, empty, null, and error states gracefully. + +------ + +## Data Model + +## Core records + +- User +- SupportConversation +- CaseReference +- CaseActionLog +- EscalationRequest +- KnowledgeArticle +- Notification +- EvalRun + +## Key fields + +- `salesforce_case_id` +- `case_number` +- `account_id` +- `contact_id` +- `priority` +- `status` +- `owner` +- `sla_due_at` +- `last_synced_at` +- `approval_required` + +------ + +## Non-functional Requirements + +- P95 tool response under 3 seconds for cached reads. +- P95 under 8 seconds for live Salesforce calls. +- Graceful fallback when Salesforce API is slow or unavailable. +- All writes must be idempotent. +- All external data should be cached with TTL where appropriate. +- Streaming UI must remain usable on mobile and desktop. + +Salesforce Developer Edition is available for testing and development, so this can be built against a real org without needing production access.developer.salesforce+1 + +------ + +## MVP Cut + +For the portfolio version, the MVP should include: + +- Login and role-based routing. +- Search Salesforce cases. +- Fetch customer context. +- Search knowledge base and similar tickets. +- Draft reply generation. +- Create/update case. +- Escalation with human approval. +- Team lead dashboard. +- Notification bell. +- Observability and tests. + +That is enough to demonstrate an FDE-grade system with real enterprise integration, agentic reasoning, and polished UX.careers.salesforce+2 + +------ + +## Out of Scope + +- Full omnichannel inbox. +- Omnichannel routing engine. +- Voice support. +- Multilingual support. +- Deep Salesforce customization beyond standard objects. +- Complex forecasting or revenue analytics. +- Mobile app. + +These can come later, but they are unnecessary for the portfolio goal. + diff --git a/apps/agent-core/.github/workflows/llm-free-tests.yml b/apps/agent-core/.github/workflows/llm-free-tests.yml new file mode 100644 index 000000000..f353756d6 --- /dev/null +++ b/apps/agent-core/.github/workflows/llm-free-tests.yml @@ -0,0 +1,25 @@ +name: LLM-Free Tests + +on: + push: + branches: [main, master, 'feat/**', 'fix/**'] + pull_request: + branches: [main, master] + +jobs: + test: + runs-on: ubuntu-latest + defaults: + run: + working-directory: apps/agent-core + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + - name: Install uv + run: pip install uv + - name: Sync dependencies + run: uv sync + - name: Run LLM-free tests + run: uv run pytest tests/test_llm_free_*.py -q --tb=short diff --git a/apps/agent-core/main.py b/apps/agent-core/main.py index 6d784df8e..cf8115c04 100644 --- a/apps/agent-core/main.py +++ b/apps/agent-core/main.py @@ -107,6 +107,9 @@ async def stream_chat(body: StreamRequest): @app.get("/health") async def health(): + is_mock = os.environ.get("LLM_PROVIDER", "cohere").lower().strip() == "mock" + if is_mock: + return {"status": "ok", "service": "agent-core", "version": "1.0.0", "postgres": False, "mock": True} pool = await get_pool() async with pool.acquire() as conn: pg_ok = bool(await conn.fetchval("SELECT 1")) diff --git a/apps/agent-core/mcp_server.py b/apps/agent-core/mcp_server.py index 34dea46ae..02656817c 100644 --- a/apps/agent-core/mcp_server.py +++ b/apps/agent-core/mcp_server.py @@ -13,7 +13,12 @@ from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent -from mcp.server.lifecycle import LifespanManager + +try: + from mcp.server.lifecycle import LifespanManager +except ImportError: + # Newer mcp versions use different lifecycle + LifespanManager = None # Import our tools from src.db import get_pool, close_pool diff --git a/apps/agent-core/migrations/006_add_support_tables.sql b/apps/agent-core/migrations/006_add_support_tables.sql new file mode 100644 index 000000000..e91d2ff76 --- /dev/null +++ b/apps/agent-core/migrations/006_add_support_tables.sql @@ -0,0 +1,146 @@ +-- Migration 006: Add SupportPilot tables for salesforce customer support cockpit +-- Spec: SupportPilot Phase 1 — Support Schema Migration +-- +-- This migration is additive. All existing procurement tables remain untouched. +-- New support tables: SupportConversation, CaseReference, EscalationRequest, +-- KnowledgeArticle, SlaPolicy. +-- +-- Run against your database: +-- psql -U supabase_admin -d postgres -f migrations/006_add_support_tables.sql +-- Or via Docker: +-- docker exec -i supabase-db psql -U supabase_admin -d postgres < migrations/006_add_support_tables.sql + +-- Ensure pgvector extension for embedding column on KnowledgeArticle +CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public; + +-- ============================================================ +-- TABLE: SupportConversation +-- Tracks customer support chat sessions / conversations +-- ============================================================ + +CREATE TABLE IF NOT EXISTS "SupportConversation" ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + title TEXT, + status TEXT NOT NULL DEFAULT 'open', + user_id TEXT REFERENCES users(id), + salesforce_case_id TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON TABLE "SupportConversation" IS 'Customer support chat sessions initiated via the SupportPilot cockpit. Each conversation tracks interaction context, status lifecycle, and links to a Salesforce case.'; +COMMENT ON COLUMN "SupportConversation".id IS 'Auto-generated UUID primary key'; +COMMENT ON COLUMN "SupportConversation".title IS 'Human-readable conversation label (e.g. "Order delay inquiry #1234")'; +COMMENT ON COLUMN "SupportConversation".status IS 'Current lifecycle state: open, pending, resolved, closed'; +COMMENT ON COLUMN "SupportConversation".user_id IS 'FK to the support agent or customer who initiated the conversation'; +COMMENT ON COLUMN "SupportConversation".salesforce_case_id IS 'Corresponding Salesforce case identifier for cross-referencing'; +COMMENT ON COLUMN "SupportConversation".created_at IS 'Timestamp of conversation creation'; +COMMENT ON COLUMN "SupportConversation".updated_at IS 'Timestamp of last conversation update'; + +-- ============================================================ +-- TABLE: CaseReference +-- Cached Salesforce case data for fast lookup without API round-trips +-- ============================================================ + +CREATE TABLE IF NOT EXISTS "CaseReference" ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + conversation_id UUID REFERENCES "SupportConversation"(id), + salesforce_case_id TEXT NOT NULL, + case_number TEXT, + subject TEXT, + status TEXT, + priority TEXT, + owner TEXT, + account_id TEXT, + contact_id TEXT, + last_synced_at TIMESTAMPTZ +); + +COMMENT ON TABLE "CaseReference" IS 'Cached snapshot of Salesforce case data. Minimizes API calls by storing frequently accessed fields locally. Refreshed on-demand via sync trigger.'; +COMMENT ON COLUMN "CaseReference".id IS 'Auto-generated UUID primary key'; +COMMENT ON COLUMN "CaseReference".conversation_id IS 'FK to the SupportConversation this case reference belongs to'; +COMMENT ON COLUMN "CaseReference".salesforce_case_id IS 'Unique Salesforce case identifier (NOT NULL — required for cross-reference)'; +COMMENT ON COLUMN "CaseReference".case_number IS 'Human-readable case number from Salesforce (e.g. 00001234)'; +COMMENT ON COLUMN "CaseReference".subject IS 'Case subject line as entered in Salesforce'; +COMMENT ON COLUMN "CaseReference".status IS 'Case status from Salesforce (New, Working, Escalated, Closed)'; +COMMENT ON COLUMN "CaseReference".priority IS 'Case priority (Low, Medium, High, Critical)'; +COMMENT ON COLUMN "CaseReference".owner IS 'Name of the assigned Salesforce case owner'; +COMMENT ON COLUMN "CaseReference".account_id IS 'Salesforce Account ID associated with this case'; +COMMENT ON COLUMN "CaseReference".contact_id IS 'Salesforce Contact ID associated with this case'; +COMMENT ON COLUMN "CaseReference".last_synced_at IS 'Timestamp of the most recent sync from Salesforce API'; + +-- ============================================================ +-- TABLE: EscalationRequest +-- Human-in-the-loop (HITL) approval requests for case escalations +-- ============================================================ + +CREATE TABLE IF NOT EXISTS "EscalationRequest" ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + case_id UUID REFERENCES "CaseReference"(id), + reason TEXT NOT NULL, + requested_action TEXT, + status TEXT NOT NULL DEFAULT 'pending', + requested_by TEXT REFERENCES users(id), + decided_by UUID, + decision TEXT, + decided_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON TABLE "EscalationRequest" IS 'Human-in-the-loop escalation requests. When the AI agent determines a case needs supervisor judgment, it creates an EscalationRequest for manual review and approval.'; +COMMENT ON COLUMN "EscalationRequest".id IS 'Auto-generated UUID primary key'; +COMMENT ON COLUMN "EscalationRequest".case_id IS 'FK to the CaseReference being escalated'; +COMMENT ON COLUMN "EscalationRequest".reason IS 'Detailed explanation of why escalation is needed (NOT NULL)'; +COMMENT ON COLUMN "EscalationRequest".requested_action IS 'Suggested action for the reviewer (e.g. "Approve refund of $250")'; +COMMENT ON COLUMN "EscalationRequest".status IS 'Escalation lifecycle: pending, approved, rejected, cancelled'; +COMMENT ON COLUMN "EscalationRequest".requested_by IS 'FK to the user who requested the escalation'; +COMMENT ON COLUMN "EscalationRequest".decided_by IS 'FK to the user who made the decision (set when status changes from pending)'; +COMMENT ON COLUMN "EscalationRequest".decision IS 'Decision notes from the reviewer'; +COMMENT ON COLUMN "EscalationRequest".decided_at IS 'Timestamp when a decision was made (null while pending)'; +COMMENT ON COLUMN "EscalationRequest".created_at IS 'Timestamp of escalation request creation'; + +-- ============================================================ +-- TABLE: KnowledgeArticle +-- RAG source documents for AI-powered support answers +-- ============================================================ + +CREATE TABLE IF NOT EXISTS "KnowledgeArticle" ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + title TEXT NOT NULL, + content TEXT NOT NULL, + category TEXT, + salesforce_article_id TEXT, + embedding vector(1536), + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON TABLE "KnowledgeArticle" IS 'Knowledge base articles used as RAG (Retrieval-Augmented Generation) source material. The AI agent retrieves relevant articles via semantic search on the embedding column to answer customer queries.'; +COMMENT ON COLUMN "KnowledgeArticle".id IS 'Auto-generated UUID primary key'; +COMMENT ON COLUMN "KnowledgeArticle".title IS 'Article title (NOT NULL — required for display in search results)'; +COMMENT ON COLUMN "KnowledgeArticle".content IS 'Full article body text (NOT NULL — the primary RAG source)'; +COMMENT ON COLUMN "KnowledgeArticle".category IS 'Article category for faceted browsing (e.g. Shipping, Returns, Billing)'; +COMMENT ON COLUMN "KnowledgeArticle".salesforce_article_id IS 'Optional Salesforce Knowledge article ID for cross-reference'; +COMMENT ON COLUMN "KnowledgeArticle".embedding IS 'OpenAI text-embedding-3-small vector (1536 dimensions) for semantic similarity search'; +COMMENT ON COLUMN "KnowledgeArticle".created_at IS 'Timestamp of article creation'; + +-- ============================================================ +-- TABLE: SlaPolicy +-- Service Level Agreement definitions mapped to case priorities +-- ============================================================ + +CREATE TABLE IF NOT EXISTS "SlaPolicy" ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + name TEXT NOT NULL, + priority TEXT NOT NULL, + response_hours INTEGER NOT NULL, + resolution_hours INTEGER NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON TABLE "SlaPolicy" IS 'SLA policy definitions that drive escalation triggers. Each policy maps a case priority to target response and resolution times. The AI agent checks SLA compliance when handling cases.'; +COMMENT ON COLUMN "SlaPolicy".id IS 'Auto-generated UUID primary key'; +COMMENT ON COLUMN "SlaPolicy".name IS 'Policy display name (e.g. "Premium Support", "Standard Support")'; +COMMENT ON COLUMN "SlaPolicy".priority IS 'Matching case priority (Critical, High, Medium, Low)'; +COMMENT ON COLUMN "SlaPolicy".response_hours IS 'Target response time in hours (integer, NOT NULL)'; +COMMENT ON COLUMN "SlaPolicy".resolution_hours IS 'Target resolution time in hours (integer, NOT NULL)'; +COMMENT ON COLUMN "SlaPolicy".created_at IS 'Timestamp of policy creation'; diff --git a/apps/agent-core/pyproject.toml b/apps/agent-core/pyproject.toml index a570e0b98..c1c5423db 100644 --- a/apps/agent-core/pyproject.toml +++ b/apps/agent-core/pyproject.toml @@ -1,3 +1,7 @@ [tool.pytest.ini_options] asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" +markers = [ + "integration: marks tests as integration tests (Docker + real LLM)", + "slow: marks tests as slow (deselect with '-m \"not slow\"')", +] diff --git a/apps/agent-core/requirements.txt b/apps/agent-core/requirements.txt index f43ee8f5a..f8591d7de 100644 --- a/apps/agent-core/requirements.txt +++ b/apps/agent-core/requirements.txt @@ -1,5 +1,6 @@ fastapi==0.115.0 uvicorn[standard]==0.34.0 +mcp>=1.0.0 langgraph==0.2.60 langchain-openai==0.3.0 langchain-core==0.3.45 @@ -12,3 +13,6 @@ python-jose[cryptography]==3.3.0 pytest==8.3.0 pytest-asyncio==0.25.0 langgraph-checkpoint-redis==0.1.0 +python-dotenv>=1.0.0 +loguru>=0.7.0 +sse-starlette>=2.0.0 diff --git a/apps/agent-core/routers/__init__.py b/apps/agent-core/routers/__init__.py new file mode 100644 index 000000000..e38ed1166 --- /dev/null +++ b/apps/agent-core/routers/__init__.py @@ -0,0 +1 @@ +# routers package \ No newline at end of file diff --git a/apps/agent-core/routers/chat.py b/apps/agent-core/routers/chat.py new file mode 100644 index 000000000..508e2e954 --- /dev/null +++ b/apps/agent-core/routers/chat.py @@ -0,0 +1,178 @@ +""" +Chat Router - POST /agent/chat (SSE streaming) +Based on AGENT_CORE_RUNNING.md documentation +""" +from fastapi import APIRouter, Request, HTTPException +from fastapi.responses import StreamingResponse +from sse_starlette.sse import EventSourceResponse +from pydantic import BaseModel +from typing import List, Optional +import json +import asyncio + +from src.graph import graph +from src.dependencies import get_llm +from loguru import logger + +router = APIRouter(prefix="/agent", tags=["chat"]) + + +class ChatMessage(BaseModel): + role: str + content: str + + +class StreamRequest(BaseModel): + messages: List[ChatMessage] + user_id: str + user_role: Optional[str] = None + thread_id: Optional[str] = None + configurable: Optional[dict] = None + + +@router.post("/chat") +async def chat(request: Request, body: StreamRequest): + """ + Chat with agent - returns SSE stream + + Auth: JWT required (validated by middleware) + Returns: text/event-stream with: + - messages/partial: AI text chunks (data: [{content, type}]) + - custom: GenUI __ui__ payloads (data: {type: "ui", name, props}) + - thread_id: conversation ID + - end: stream complete + - error: error message + """ + # Check auth header (allow test mode without auth) + test_mode = request.headers.get("x-test-mode") == "true" or request.headers.get("x-user-id") == "test-user-id" + if not test_mode: + auth = request.headers.get("authorization") + if not auth: + raise HTTPException(status_code=401, detail="Unauthorized") + + try: + # Convert messages to LangChain format + from langchain_core.messages import HumanMessage, AIMessage + + langchain_messages = [] + for msg in body.messages: + if msg.role == "user": + langchain_messages.append(HumanMessage(content=msg.content)) + elif msg.role == "assistant": + langchain_messages.append(AIMessage(content=msg.content)) + + # Build initial state — user_role can come from body.user_role + # or configurable.role (in order of precedence) + effective_role = ( + body.user_role + or (body.configurable.get("role") if body.configurable else None) + or "EMPLOYEE" + ) + initial_state = { + "messages": langchain_messages, + "user_id": body.user_id, + "user_role": effective_role, + "step_count": 0, + } + + # Stream the graph execution + async def event_generator(): + try: + async for event in graph.astream(initial_state, stream_mode="values"): + # Check for messages + if "messages" in event: + last_msg = event["messages"][-1] + if hasattr(last_msg, "content") and last_msg.content: + raw_content = last_msg.content + ui_payload = None + display_content = None + + # Parse structured JSON content that embeds + # __ui__ alongside the text content. + # + # This handles TWO paths: + # 1. Real LLM: ToolNode returns ToolMessage whose + # content is JSON with __ui__ + content fields + # 2. MockLLM/any provider: AIMessage whose content + # is JSON wrapping __ui__ + native text + try: + parsed = json.loads(raw_content) + if isinstance(parsed, dict): + # Pop __ui__ so it doesn't leak to text + ui_payload = parsed.pop("__ui__", None) + # Use clean content field if available + display_content = parsed.get("content", "") + except (json.JSONDecodeError, TypeError): + # Not JSON — use as plain text + pass + + # Fall back to raw content if no cleaner version + if display_content is None: + display_content = raw_content + + # Canonical SSE format (src/sse.py spec) + if display_content: + yield { + "event": "messages/partial", + "data": json.dumps( + [{"content": display_content, "type": "ai"}] + ), + } + # Backward-compatible delta (legacy chat pages) + yield { + "event": "delta", + "data": json.dumps({"content": display_content}), + } + + # Emit UI payload if found in message content + if ui_payload: + yield { + "event": "custom", + "data": json.dumps({"type": "ui", **ui_payload}), + } + # Backward-compatible ui_actions + yield { + "event": "ui_actions", + "data": json.dumps({"actions": [ui_payload]}), + } + + # Check for UI actions in last_tool_result (backup path) + if "last_tool_result" in event and event["last_tool_result"]: + result = event["last_tool_result"] + if "__ui__" in result: + ui = result["__ui__"] + # Canonical SSE format + yield { + "event": "custom", + "data": json.dumps({"type": "ui", **ui}), + } + # Backward-compatible ui_actions (legacy chat pages) + yield { + "event": "ui_actions", + "data": json.dumps({"actions": [ui]}), + } + + # Check for pending PR (approval flow) + if "pending_pr_id" in event and event["pending_pr_id"]: + yield {"event": "thread_id", "data": json.dumps({"threadId": event.get("pending_pr_number", "")})} + + # Canonical end event + yield {"event": "end", "data": json.dumps({})} + # Backward-compatible complete event + yield {"event": "complete", "data": json.dumps({})} + + except Exception as e: + logger.error(f"Chat error: {e}") + yield {"event": "error", "data": json.dumps({"message": str(e)})} + + return EventSourceResponse(event_generator()) + + except Exception as e: + logger.error(f"Chat endpoint error: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/stream") +async def stream(request: Request, body: StreamRequest): + """Alternative streaming endpoint - same as /chat""" + return await chat(request, body) \ No newline at end of file diff --git a/apps/agent-core/src/dependencies.py b/apps/agent-core/src/dependencies.py index af0783d35..9ea8038e2 100644 --- a/apps/agent-core/src/dependencies.py +++ b/apps/agent-core/src/dependencies.py @@ -4,7 +4,6 @@ """ import asyncpg import redis.asyncio as aioredis -from langchain_openai import ChatOpenAI from contextlib import asynccontextmanager from fastapi import FastAPI import os @@ -24,52 +23,90 @@ # ── Module-level singletons ────────────────────── _db_pool: asyncpg.Pool | None = None _redis: aioredis.Redis | None = None -_llm: ChatOpenAI | None = None +_llm: "Any | None" = None _langfuse: Langfuse | None = None +_salesforce_client: "MockSalesforceClient | None" = None + + +def init_salesforce_client( + mode: str | None = None, + api_key: str | None = None, + instance_url: str | None = None, +): + """Initialize the global Salesforce client singleton. + + Args: + mode: 'mock' (default) or 'live'. Falls back to SALESFORCE_MODE env var. + api_key: Optional Salesforce API key. Falls back to SALESFORCE_API_KEY env var. + instance_url: Optional Salesforce instance URL. Falls back to SALESFORCE_INSTANCE_URL env var. + """ + global _salesforce_client + from src.salesforce import MockSalesforceClient + + _salesforce_client = MockSalesforceClient( + mode=mode or os.environ.get("SALESFORCE_MODE", "mock"), + api_key=api_key or os.environ.get("SALESFORCE_API_KEY"), + instance_url=instance_url or os.environ.get("SALESFORCE_INSTANCE_URL"), + ) + print(f"✅ Salesforce client initialized (mode={_salesforce_client.mode})") + return _salesforce_client -@asynccontextmanager -async def lifespan(app: FastAPI): - """Initialize all clients ONCE at startup.""" - global _db_pool, _redis, _llm +def get_salesforce_client(): + """Get the global Salesforce client singleton.""" + return _salesforce_client - database_url = os.environ.get("DATABASE_URL") - if not database_url: - raise RuntimeError("DATABASE_URL not set") - _db_pool = await asyncpg.create_pool( - database_url, - min_size=2, - max_size=10, - command_timeout=60, - ) - print(f"✅ DB pool initialized: {database_url}") +def shutdown_salesforce_client(): + """Clean up Salesforce client on shutdown.""" + global _salesforce_client + _salesforce_client = None + print("✅ Salesforce client shutdown") + - redis_url = os.environ.get("REDIS_URL") - if redis_url: - _redis = await aioredis.from_url( - redis_url, - decode_responses=True, +@asynccontextmanager +async def lifespan(app: FastAPI): + """Initialize all clients ONCE at startup.""" + global _db_pool, _redis, _llm, _salesforce_client + + provider = os.environ.get("LLM_PROVIDER", "cohere").lower().strip() + is_mock = provider == "mock" + print(f"🔧 LLM_PROVIDER={provider}") + + if not is_mock: + database_url = os.environ.get("DATABASE_URL") + if not database_url: + raise RuntimeError("DATABASE_URL not set") + + _db_pool = await asyncpg.create_pool( + database_url, + min_size=2, + max_size=10, + command_timeout=60, ) - print(f"✅ Redis initialized: {redis_url}") + print(f"✅ DB pool initialized") - llm_model = os.environ.get("OLLAMA_MODEL", "nemotron-3-super:cloud") - llm_base_url = os.environ.get("OLLAMA_BASE_URL", "http://localhost:11434/v1") - llm_api_key = os.environ.get("OLLAMA_API_KEY", "ollama") + redis_url = os.environ.get("REDIS_URL") + if redis_url: + _redis = await aioredis.from_url( + redis_url, + decode_responses=True, + ) + print(f"✅ Redis initialized") - _llm = ChatOpenAI( - model=llm_model, - temperature=0, - base_url=llm_base_url, - api_key=llm_api_key, - ) - print(f"✅ LLM initialized: {llm_model}") + from src.llm_config import create_llm + + _llm = create_llm() if LANGFUSE_AVAILABLE: global _langfuse _langfuse = Langfuse() print("✅ Langfuse initialized") + # Initialize Salesforce client (always, even in mock mode) + init_salesforce_client() + print("✅ Salesforce client initialized") + yield # ← app runs here # Clean shutdown @@ -79,6 +116,7 @@ async def lifespan(app: FastAPI): if _redis: await _redis.aclose() print("✅ Redis closed") + shutdown_salesforce_client() def get_pool() -> asyncpg.Pool: @@ -93,7 +131,7 @@ def get_redis() -> aioredis.Redis: return _redis -def get_llm() -> ChatOpenAI: +def get_llm() -> "Any": if _llm is None: raise RuntimeError("LLM not initialized - ensure lifespan is used") return _llm @@ -105,19 +143,18 @@ def get_langfuse() -> "Langfuse | None": def get_langfuse_metadata(config: dict = None) -> dict: - """Get Langfuse metadata from config for tracing (PRD Part 9).""" + """Get Langfuse metadata from config for tracing.""" if not config: - return {"app": "procureai"} + return {"app": "supportpilot"} - # Handle non-dict config - return default app only if not isinstance(config, dict): - return {"app": "procureai"} + return {"app": "supportpilot"} cfg = config.get("configurable", {}) if isinstance(config, dict) else {} return { "department_id": cfg.get("department_id", "unknown"), "role": cfg.get("role", "unknown"), - "app": "procureai", + "app": "supportpilot", } diff --git a/apps/agent-core/src/eval_suite.py b/apps/agent-core/src/eval_suite.py new file mode 100644 index 000000000..052c4b9ed --- /dev/null +++ b/apps/agent-core/src/eval_suite.py @@ -0,0 +1,169 @@ +""" +Eval Test Suite for SupportPilot — LLM-as-judge evaluation for support agent responses. + +Covers support-specific failure modes: +- Wrong tool selection (calling escalate as SUPPORT_AGENT) +- Case creation before search (creating duplicates) +- Missing customer context lookup +- GenUI null fields +- Context confusion +""" + +from typing import Any + + +# Evaluation cases for support-only agent +EVAL_CASES = [ + { + "name": "agent_cannot_escalate", + "input": "Escalate case 500ABC to urgent", + "expected_tools": ["escalate_case"], + "must_not_call": ["escalate_case"], + "role": "SUPPORT_AGENT", + "expected_outcome": "error — role insufficient", + "failure_mode": "wrong_tool_selection", + }, + { + "name": "team_lead_can_escalate", + "input": "Escalate case 500ABC to urgent", + "expected_tools": ["escalate_case"], + "role": "TEAM_LEAD", + "expected_outcome": "case_escalated", + "failure_mode": "none", + }, + { + "name": "search_before_create", + "input": "I have a billing issue with my account", + "expected_tools": ["search_salesforce_cases"], + "must_not_call": ["create_case"], + "role": "SUPPORT_AGENT", + "expected_outcome": "search_performed", + "failure_mode": "wrong_tool_selection", + }, + { + "name": "get_customer_context_on_query", + "input": "What's the status for Acme Corp?", + "expected_tools": ["get_customer_context"], + "role": "SUPPORT_AGENT", + "expected_outcome": "context_loaded", + "failure_mode": "context_confusion", + }, + { + "name": "kb_search_for_troubleshooting", + "input": "How do I reset a user password?", + "expected_tools": ["search_knowledge_base"], + "role": "SUPPORT_AGENT", + "expected_outcome": "kb_searched", + "failure_mode": "none", + }, + { + "name": "case_details_by_number", + "input": "Show me case 500ABC", + "expected_tools": ["get_case_details"], + "role": "SUPPORT_AGENT", + "expected_outcome": "case_details_shown", + "failure_mode": "none", + }, + { + "name": "draft_reply_for_case", + "input": "Help me write a reply to case 500ABC about the refund", + "expected_tools": ["draft_case_reply"], + "role": "SUPPORT_AGENT", + "expected_outcome": "draft_generated", + "failure_mode": "none", + }, + { + "name": "create_case_new_issue", + "input": "Open a new case for login issue with jane@acme.com", + "expected_tools": ["create_case"], + "role": "SUPPORT_AGENT", + "expected_outcome": "case_created", + "failure_mode": "none", + }, + { + "name": "update_case_status", + "input": "Mark case 500ABC as resolved", + "expected_tools": ["update_case"], + "role": "SUPPORT_AGENT", + "expected_outcome": "case_updated", + "failure_mode": "none", + }, + { + "name": "support_ops_read_only", + "input": "Search for cases related to billing", + "expected_tools": ["search_salesforce_cases"], + "must_not_call": ["create_case", "update_case", "escalate_case"], + "role": "SUPPORT_OPS", + "expected_outcome": "search_only", + "failure_mode": "wrong_tool_selection", + }, +] + + +def evaluate_response( + user_input: str, + role: str, + tool_calls: list[dict[str, Any]] | None = None, + expected_outcome: str | None = None, + ui_response: dict | None = None, +) -> dict[str, Any]: + """ + Evaluate an agent response against expected behavior. + + Returns a dict with: + - passed: bool + - reason: str explanation + - failure_mode: str if failed + """ + tool_names = [tc["name"] for tc in (tool_calls or [])] + + # Check: SUPPORT_AGENT calling escalate_case should be blocked + if role == "SUPPORT_AGENT" and "escalate_case" in tool_names: + return { + "passed": False, + "reason": "blocked_role_insufficient", + "failure_mode": "wrong_tool_selection", + } + + # Check: SUPPORT_OPS calling create/update/escalate should be blocked + if role == "SUPPORT_OPS": + blocked = {"create_case", "update_case", "escalate_case"} + if any(t in tool_names for t in blocked): + return { + "passed": False, + "reason": "support_ops_read_only_restricted", + "failure_mode": "wrong_tool_selection", + } + + # Check: Creating a case without searching first + if "create_case" in tool_names and "search_salesforce_cases" not in tool_names: + # Only flag if the input suggests existing issue tracking + search_keywords = ["issue", "problem", "not working", "broken", "error"] + search_keywords += ["bug", "fail", "down", "cannot"] + if any(kw in user_input.lower() for kw in search_keywords): + return { + "passed": False, + "reason": "created_case_without_search", + "failure_mode": "wrong_tool_selection", + } + + # Check: GenUI null fields + if ui_response: + for key, value in ui_response.items(): + if isinstance(value, list): + for item in value: + if isinstance(item, dict): + for field_key, field_val in item.items(): + if field_val is None and field_key in ("id", "caseNumber", "status"): + return { + "passed": False, + "reason": f"null_{field_key}_in_ui", + "failure_mode": "genui_null_crash", + } + + # Default: pass + return { + "passed": True, + "reason": "expected_behavior", + "failure_mode": "none", + } diff --git a/apps/agent-core/src/graph.py b/apps/agent-core/src/graph.py index 4db5a731d..a0d38a8c8 100644 --- a/apps/agent-core/src/graph.py +++ b/apps/agent-core/src/graph.py @@ -1,20 +1,39 @@ -import os import json -from typing import Annotated, TypedDict, Optional, Literal -from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage +from typing import Annotated, TypedDict, Optional +from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, ToolMessage from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode -from langgraph.types import interrupt, Command from loguru import logger -from .tools import ALL_TOOLS +from .tools import ALL_TOOLS, get_tools_for_role -logger.add( - "/tmp/agent.log", - rotation="10 MB", - level="DEBUG", - format="- 👋 Hi{session?.user?.name ? `, ${session.user.name}` : ''}! How can I help? -
-