diff --git a/.env.example b/.env.example index ff0bc6e..31e81fb 100644 --- a/.env.example +++ b/.env.example @@ -29,6 +29,11 @@ RK_KNOWLEDGE_BASES_AUDIENCE=kb-service RK_KNOWLEDGE_BASES_JWT_ALGORITHM=HS256 RK_KNOWLEDGE_BASES_TIMEOUT_SECONDS=20 RK_KNOWLEDGE_BASES_CONNECT_TIMEOUT_SECONDS=5 +# Service-to-service auth for dashboard `api` -> agent internal endpoints (/internal/*). +# Must match RK_INTERNAL_SERVICE_SIGNING_KEY on the dashboard side. +RK_INTERNAL_SERVICE_SIGNING_KEY=change-me-internal-service-signing-key +RK_INTERNAL_SERVICE_AUDIENCE=agent-service +RK_INTERNAL_SERVICE_JWT_ALGORITHM=HS256 # Frontend NEXT_PUBLIC_API_BASE_URL=http://localhost:3002 diff --git a/agent/config.py b/agent/config.py index ab96abd..f6dd05e 100644 --- a/agent/config.py +++ b/agent/config.py @@ -43,5 +43,9 @@ class Settings(BaseSettings): knowledge_bases_timeout_seconds: float = 20.0 knowledge_bases_connect_timeout_seconds: float = 5.0 + internal_service_signing_key: str = "change-me-internal-service-signing-key" + internal_service_audience: str = "agent-service" + internal_service_jwt_algorithm: str = "HS256" + settings = Settings() diff --git a/agent/middleware/internal_auth.py b/agent/middleware/internal_auth.py new file mode 100644 index 0000000..047bb43 --- /dev/null +++ b/agent/middleware/internal_auth.py @@ -0,0 +1,24 @@ +from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +import jwt +from jwt import InvalidTokenError + +from agent.config import settings + +bearer_scheme = HTTPBearer() + + +async def require_internal_service( + credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme), +) -> None: + """Validates a service-to-service JWT issued by the dashboard `api`, mirroring + the knowledge_bases service's own internal-auth pattern.""" + try: + jwt.decode( + credentials.credentials, + settings.internal_service_signing_key, + algorithms=[settings.internal_service_jwt_algorithm], + audience=settings.internal_service_audience, + ) + except InvalidTokenError as exc: + raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid service token") from exc diff --git a/agent/models/__init__.py b/agent/models/__init__.py index e3e896b..dee50ad 100644 --- a/agent/models/__init__.py +++ b/agent/models/__init__.py @@ -12,6 +12,7 @@ from agent.models.knowledge_base_ref import KnowledgeBaseRef from agent.models.session import ChatSession from agent.models.message import Message +from agent.models.session_feedback import SessionFeedback from agent.models.llm_usage_event import LLMUsageEvent from agent.models.playbook import Playbook, PlaybookFunction @@ -30,6 +31,7 @@ "KnowledgeBaseRef", "ChatSession", "Message", + "SessionFeedback", "LLMUsageEvent", "Playbook", "PlaybookFunction", diff --git a/agent/models/session.py b/agent/models/session.py index c5a26a1..326ad14 100644 --- a/agent/models/session.py +++ b/agent/models/session.py @@ -2,7 +2,7 @@ from datetime import datetime, timezone from typing import TYPE_CHECKING, Any -from sqlalchemy import DateTime, ForeignKey, String +from sqlalchemy import DateTime, ForeignKey, String, Text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship @@ -11,6 +11,7 @@ if TYPE_CHECKING: from agent.models.app import App from agent.models.message import Message + from agent.models.session_feedback import SessionFeedback class ChatSession(Base, UUIDMixin, TimestampMixin): @@ -18,7 +19,7 @@ class ChatSession(Base, UUIDMixin, TimestampMixin): app_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("apps.id", ondelete="CASCADE")) device_id: Mapped[str | None] = mapped_column(String(255)) - status: Mapped[str] = mapped_column(String(20), default="active") # active, expired, closed + status: Mapped[str] = mapped_column(String(20), default="active") # active, expired, closed, escalated last_activity_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) ) @@ -26,6 +27,10 @@ class ChatSession(Base, UUIDMixin, TimestampMixin): llm_context: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict) available_function_names: Mapped[list[str]] = mapped_column(JSONB, default=list) locale: Mapped[str] = mapped_column(String(16), default="en") + escalated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + escalation_reason: Mapped[str | None] = mapped_column(Text, nullable=True) + resolved_by: Mapped[str | None] = mapped_column(String(20), nullable=True) # ai, human app: Mapped["App"] = relationship(back_populates="sessions") messages: Mapped[list["Message"]] = relationship(back_populates="session", cascade="all, delete-orphan", order_by="Message.sequence_number") + feedback: Mapped["SessionFeedback | None"] = relationship(back_populates="session", cascade="all, delete-orphan", uselist=False) diff --git a/agent/models/session_feedback.py b/agent/models/session_feedback.py new file mode 100644 index 0000000..f7bd06d --- /dev/null +++ b/agent/models/session_feedback.py @@ -0,0 +1,25 @@ +import uuid +from typing import TYPE_CHECKING + +from sqlalchemy import CheckConstraint, ForeignKey, Integer, Text +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from agent.models.base import Base, TimestampMixin, UUIDMixin + +if TYPE_CHECKING: + from agent.models.session import ChatSession + + +class SessionFeedback(Base, UUIDMixin, TimestampMixin): + __tablename__ = "session_feedback" + __table_args__ = ( + CheckConstraint("rating >= 1 AND rating <= 5", name="ck_session_feedback_rating_range"), + ) + + session_id: Mapped[uuid.UUID] = mapped_column( + ForeignKey("chat_sessions.id", ondelete="CASCADE"), unique=True + ) + rating: Mapped[int] = mapped_column(Integer) + comment: Mapped[str | None] = mapped_column(Text, nullable=True) + + session: Mapped["ChatSession"] = relationship(back_populates="feedback") diff --git a/agent/routers/chat_events.py b/agent/routers/chat_events.py index 133834d..2e47e1b 100644 --- a/agent/routers/chat_events.py +++ b/agent/routers/chat_events.py @@ -5,6 +5,8 @@ import uuid from typing import Any +import redis.exceptions as redis_exceptions + logger = logging.getLogger(__name__) from fastapi import APIRouter, Depends, HTTPException, Request, status @@ -15,8 +17,10 @@ from agent.database import async_session_factory, get_db from agent.middleware.auth import get_app_from_sdk_auth +from agent.middleware.internal_auth import require_internal_service from agent.models.agent_config import AgentConfig from agent.models.app import App +from agent.models.message import Message from agent.models.organization_llm_provider_profile import OrganizationLLMProviderProfile from agent.models.session import ChatSession from agent.services.chat_access_service import ( @@ -33,7 +37,7 @@ register_pending_tool_result, resolve_pending_tool_result, ) -from agent.services.session_service import is_session_expired +from agent.services.session_service import get_next_sequence, is_session_expired from agent.services.turn_state_service import turn_state_store router = APIRouter(tags=["chat-events"]) @@ -63,6 +67,14 @@ class ChatMessageAccepted(BaseModel): status: str = "accepted" +class HumanMessageBody(BaseModel): + text: str = Field(min_length=1, max_length=_MAX_MESSAGE_TEXT_BYTES) + + +class SessionEscalatedBody(BaseModel): + reason: str = Field(min_length=1, max_length=2000) + + class ToolResultBody(BaseModel): turn_id: str idempotency_key: str = Field(min_length=1, max_length=255) @@ -131,9 +143,15 @@ async def send_tool_call_request( async def send_turn_complete(self, full_text: str, usage: dict | None) -> None: await self._push("turn_complete", {"full_text": full_text, "usage": usage}) + async def send_feedback_requested(self) -> None: + await self._push("feedback_requested", {"immediate": False}) + async def send_error(self, code: str, message: str, recoverable: bool = True) -> None: await self._push("error", {"code": code, "message": message, "recoverable": recoverable}) + async def send_session_escalated(self, reason: str) -> None: + await self._push("session_escalated", {"reason": reason}) + async def wait_for_tool_result(self, call_id: str, timeout: int) -> dict[str, Any]: fut = self._pending.get(call_id) if not fut: @@ -248,6 +266,14 @@ async def generate(): except TimeoutError: yield ": keep-alive\n\n" continue + except redis_exceptions.RedisError: + # Transient Redis hiccup (e.g. a blocking BGSAVE) — degrade to a + # keep-alive and retry rather than crashing the whole SSE stream. + logger.warning( + "sse_redis_hiccup session_id=%s app_id=%s", session.id, app.id, exc_info=True + ) + yield ": keep-alive\n\n" + continue if not events: yield ": keep-alive\n\n" @@ -340,3 +366,84 @@ async def submit_tool_result( raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No pending tool call with this ID") return {"status": "ok", "deduplicated": False} + + +@router.post("/internal/sessions/{session_id}/human-message", dependencies=[Depends(require_internal_service)]) +async def post_human_message( + session_id: uuid.UUID, + body: HumanMessageBody, + db: AsyncSession = Depends(get_db), +): + session = await db.get(ChatSession, session_id) + if session is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found") + + seq = await get_next_sequence(db, session.id) + message = Message( + session_id=session.id, + sequence_number=seq, + role="human_agent", + content=body.text, + ) + db.add(message) + await db.commit() + await db.refresh(message) + + await event_stream_store.append( + session_id=session.id, + app_id=session.app_id, + turn_id=str(uuid.uuid4()), + request_id=str(uuid.uuid4()), + event_type="human_message", + payload={"message_id": str(message.id), "text": body.text}, + ) + + return { + "id": str(message.id), + "created_at": message.created_at.isoformat(), + "session_id": str(message.session_id), + "sequence_number": message.sequence_number, + "role": message.role, + "content": message.content, + } + + +@router.post("/internal/sessions/{session_id}/feedback-requested", dependencies=[Depends(require_internal_service)]) +async def post_feedback_requested( + session_id: uuid.UUID, + db: AsyncSession = Depends(get_db), +): + session = await db.get(ChatSession, session_id) + if session is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found") + + await event_stream_store.append( + session_id=session.id, + app_id=session.app_id, + turn_id=str(uuid.uuid4()), + request_id=str(uuid.uuid4()), + event_type="feedback_requested", + payload={"immediate": True}, + ) + + +@router.post("/internal/sessions/{session_id}/session-escalated", dependencies=[Depends(require_internal_service)]) +async def post_session_escalated( + session_id: uuid.UUID, + body: SessionEscalatedBody, + db: AsyncSession = Depends(get_db), +): + session = await db.get(ChatSession, session_id) + if session is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found") + + await event_stream_store.append( + session_id=session.id, + app_id=session.app_id, + turn_id=str(uuid.uuid4()), + request_id=str(uuid.uuid4()), + event_type="session_escalated", + payload={"reason": body.reason}, + ) + + return {"status": "ok"} diff --git a/agent/routers/sessions.py b/agent/routers/sessions.py index 953c4ed..b91fd24 100644 --- a/agent/routers/sessions.py +++ b/agent/routers/sessions.py @@ -10,11 +10,14 @@ from agent.models.app import App from agent.models.message import Message from agent.models.session import ChatSession +from agent.models.session_feedback import SessionFeedback from agent.schemas.session import ( MessageOut, SessionContextOut, SessionContextPatch, SessionCreate, + SessionFeedbackCreate, + SessionFeedbackOut, SessionOut, ) from agent.services.chat_localization_service import effective_texts, resolve_locale @@ -195,7 +198,38 @@ async def get_session_messages_sdk( result = await db.execute( select(Message).where( Message.session_id == session_id, - Message.role.in_(["user", "assistant"]), + Message.role.in_(["user", "assistant", "human_agent"]), ).order_by(Message.sequence_number) ) return result.scalars().all() + + +@sdk_router.post("/{session_id}/feedback", response_model=SessionFeedbackOut, status_code=status.HTTP_201_CREATED) +async def submit_session_feedback( + session_id: uuid.UUID, + body: SessionFeedbackCreate, + request: Request, + app: App = Depends(get_app_from_sdk_auth), + db: AsyncSession = Depends(get_db), +): + validate_chat_capability_token( + token=resolve_chat_capability_token(request.headers), + session_id=session_id, + app=app, + ) + + session = await db.get(ChatSession, session_id) + if not session or session.app_id != app.id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found") + + existing = await db.execute( + select(SessionFeedback).where(SessionFeedback.session_id == session_id) + ) + if existing.scalar_one_or_none() is not None: + raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Feedback already submitted for this session") + + feedback = SessionFeedback(session_id=session_id, rating=body.rating, comment=body.comment) + db.add(feedback) + await db.commit() + await db.refresh(feedback) + return feedback diff --git a/agent/schemas/session.py b/agent/schemas/session.py index 2ad8303..529c252 100644 --- a/agent/schemas/session.py +++ b/agent/schemas/session.py @@ -118,6 +118,23 @@ class MessageOut(BaseModel): model_config = {"from_attributes": True} +class SessionFeedbackCreate(BaseModel): + rating: int = Field(ge=1, le=5) + comment: str | None = Field(default=None, max_length=2000) + + model_config = {"extra": "forbid"} + + +class SessionFeedbackOut(BaseModel): + id: uuid.UUID + session_id: uuid.UUID + rating: int + comment: str | None + created_at: datetime + + model_config = {"from_attributes": True} + + class SessionContextOut(BaseModel): id: uuid.UUID app_id: uuid.UUID diff --git a/agent/services/orchestrator.py b/agent/services/orchestrator.py index 6a611b2..8c67b79 100644 --- a/agent/services/orchestrator.py +++ b/agent/services/orchestrator.py @@ -4,6 +4,7 @@ import re import uuid from dataclasses import dataclass +from datetime import datetime, timezone from typing import Any from sqlalchemy import select @@ -38,6 +39,7 @@ logger = logging.getLogger(__name__) KB_SEARCH_TOOL_NAME = "kb_search" +ESCALATE_TOOL_NAME = "escalate_to_human" KB_PREFETCH_MAX_ITEMS = 5 KB_PREFETCH_MAX_CHARS = 1200 KB_INDEX_MAX_ITEMS = 8 @@ -859,6 +861,39 @@ def _build_kb_search_tool() -> dict[str, Any]: } +def _build_escalate_tool() -> dict[str, Any]: + return { + "type": "function", + "function": { + "name": ESCALATE_TOOL_NAME, + "description": ( + "Escalate this conversation to a human support agent. Use this when the user " + "explicitly asks for a human, or when you cannot resolve the issue yourself." + ), + "parameters": { + "type": "object", + "properties": { + "reason": {"type": "string", "description": "Why this conversation needs a human."}, + }, + "required": ["reason"], + }, + }, + } + + +async def escalate_session( + db: AsyncSession, + session: ChatSession, + sender: "MessageSender", + reason: str, +) -> None: + session.status = "escalated" + session.escalated_at = datetime.now(timezone.utc) + session.escalation_reason = reason + await db.commit() + await sender.send_session_escalated(reason) + + async def _load_kb_assignment_context( db: AsyncSession, app_id: uuid.UUID, @@ -951,12 +986,18 @@ async def send_tool_call_request( async def send_turn_complete(self, full_text: str, usage: dict | None) -> None: raise NotImplementedError + async def send_feedback_requested(self) -> None: + raise NotImplementedError + async def send_error(self, code: str, message: str, recoverable: bool = True) -> None: raise NotImplementedError async def wait_for_tool_result(self, call_id: str, timeout: int) -> dict[str, Any]: raise NotImplementedError + async def send_session_escalated(self, reason: str) -> None: + raise NotImplementedError + async def build_playbook_prompt( db: AsyncSession, @@ -1139,6 +1180,7 @@ async def run_agent_loop( tools = list(sdk_tools) if kb_service_enabled: tools.append(_build_kb_search_tool()) + tools.append(_build_escalate_tool()) tools_payload = tools or None tool_round = 0 @@ -1240,6 +1282,7 @@ async def run_agent_loop( db.add(assistant_msg) await db.commit() await sender.send_turn_complete(accumulated_text, usage_data) + await sender.send_feedback_requested() return # 5. If tool calls → send to iOS and wait for results @@ -1271,6 +1314,7 @@ async def run_agent_loop( "arguments": arguments, "description": (fn.description_override or fn.description) if fn else "", "is_kb_internal": fn_name == KB_SEARCH_TOOL_NAME, + "is_escalation": fn_name == ESCALATE_TOOL_NAME, }) # 5a. Execute internal KB tools server-side. @@ -1314,8 +1358,24 @@ async def run_agent_loop( db.add(result_msg) await db.commit() + # 5a2. Handle escalation to human, server-side. + escalation_info = next((info for info in tc_infos if info["is_escalation"]), None) + if escalation_info is not None: + reason = str(escalation_info["arguments"].get("reason", "")).strip() or "Escalated by assistant" + seq = await get_next_sequence(db, session.id) + result_msg = Message( + session_id=session.id, + sequence_number=seq, + role="tool_result", + tool_call_id=escalation_info["id"], + content=json.dumps({"status": "escalated"}), + ) + db.add(result_msg) + await escalate_session(db, session, sender, reason) + return + # 5b. Send SDK function tools to iOS - sdk_tc_infos = [info for info in tc_infos if not info["is_kb_internal"]] + sdk_tc_infos = [info for info in tc_infos if not info["is_kb_internal"] and not info["is_escalation"]] descriptions = ( await generate_tool_descriptions(config, sdk_tc_infos, session_id=session.id) if sdk_tc_infos @@ -1406,6 +1466,29 @@ async def run_agent_loop( ) db.add(timeout_msg) await db.commit() + except Exception: + # Any other failure while waiting for the tool result (e.g. a + # transient Redis error) must not silently abort the whole turn: + # that would release the turn lock via _run_turn's `finally` + # while leaving this call permanently unresolved, letting a + # conflicting concurrent turn start for the same session. + logger.exception( + "tool_call_wait_failed session_id=%s app_id=%s call_id=%s fn=%s", + session.id, + session.app_id, + info["id"], + fn_name, + ) + seq = await get_next_sequence(db, session.id) + error_msg = Message( + session_id=session.id, + sequence_number=seq, + role="tool_result", + tool_call_id=info["id"], + content=json.dumps({"error": f"Function '{fn_name}' failed due to a transient server error"}), + ) + db.add(error_msg) + await db.commit() tool_round += 1 continue @@ -1413,6 +1496,6 @@ async def run_agent_loop( # No text and no tool calls — shouldn't happen but break to be safe break - # Max tool rounds exceeded — force text response + # Max tool rounds exceeded — escalate to a human rather than dead-ending the session if tool_round >= config.max_tool_rounds: - await sender.send_error("max_tool_rounds", "Maximum tool calling rounds exceeded", recoverable=False) + await escalate_session(db, session, sender, "Maximum tool calling rounds exceeded") diff --git a/agent/services/pending_tool_results.py b/agent/services/pending_tool_results.py index 682b030..71b4afb 100644 --- a/agent/services/pending_tool_results.py +++ b/agent/services/pending_tool_results.py @@ -156,13 +156,18 @@ async def resolve_pending_tool_result( result: dict[str, Any], ) -> bool: """Write result to Redis + publish notification + try local resolve.""" - # 1. Try Redis persistence for cross-process delivery + # 1. Try Redis persistence for cross-process delivery. Best-effort: a + # transient Redis error here must not prevent the same-process local + # future resolve below, or a real tool result would be silently dropped. redis = await get_redis_client() if redis is not None and redis_enabled(): - result_key = f"rk:tool_result:{app_id}:{session_id}:{call_id}" - await redis.set(result_key, json.dumps(result), ex=600) - channel_name = f"rk:tool_notify:{app_id}:{session_id}:{call_id}" - await redis.publish(channel_name, "ready") + try: + result_key = f"rk:tool_result:{app_id}:{session_id}:{call_id}" + await redis.set(result_key, json.dumps(result), ex=600) + channel_name = f"rk:tool_notify:{app_id}:{session_id}:{call_id}" + await redis.publish(channel_name, "ready") + except Exception: + logger.warning("tool_result_redis_publish_failed call_id=%s", call_id, exc_info=True) # 2. Try local future resolve (same-process fast path) key = _pending_key(session_id, app_id, call_id) @@ -174,8 +179,12 @@ async def resolve_pending_tool_result( # 3. If no local future, check if there's an active turn in Redis if redis is not None and redis_enabled(): - lock_key = f"rk:turn:lock:{app_id}:{session_id}" - has_turn = await redis.exists(lock_key) - return has_turn > 0 + try: + lock_key = f"rk:turn:lock:{app_id}:{session_id}" + has_turn = await redis.exists(lock_key) + return has_turn > 0 + except Exception: + logger.warning("tool_result_redis_lock_check_failed call_id=%s", call_id, exc_info=True) + return False return False diff --git a/agent/services/session_service.py b/agent/services/session_service.py index 608504c..a8a652a 100644 --- a/agent/services/session_service.py +++ b/agent/services/session_service.py @@ -90,7 +90,7 @@ async def expire_stale_sessions(db: AsyncSession) -> int: ChatSession.status == "active", ChatSession.last_activity_at < func.now() - AgentConfig.session_ttl_minutes * text("interval '1 minute'"), ) - .values(status="expired") + .values(status="expired", resolved_by="ai") ) await db.commit() return result.rowcount diff --git a/alembic/versions/024_add_session_escalation_fields.py b/alembic/versions/024_add_session_escalation_fields.py new file mode 100644 index 0000000..ef7aca5 --- /dev/null +++ b/alembic/versions/024_add_session_escalation_fields.py @@ -0,0 +1,28 @@ +"""Add escalation fields to chat_sessions + +Revision ID: 024 +Revises: 023 +Create Date: 2026-07-21 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "024" +down_revision: Union[str, None] = "023" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("chat_sessions", sa.Column("escalated_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("chat_sessions", sa.Column("escalation_reason", sa.Text(), nullable=True)) + op.add_column("chat_sessions", sa.Column("resolved_by", sa.String(length=20), nullable=True)) + + +def downgrade() -> None: + op.drop_column("chat_sessions", "resolved_by") + op.drop_column("chat_sessions", "escalation_reason") + op.drop_column("chat_sessions", "escalated_at") diff --git a/alembic/versions/025_add_session_feedback.py b/alembic/versions/025_add_session_feedback.py new file mode 100644 index 0000000..3de8265 --- /dev/null +++ b/alembic/versions/025_add_session_feedback.py @@ -0,0 +1,39 @@ +"""Add session_feedback table for CSAT capture + +Revision ID: 025 +Revises: 024 +Create Date: 2026-07-21 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "025" +down_revision: Union[str, None] = "024" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "session_feedback", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True), + sa.Column( + "session_id", + postgresql.UUID(as_uuid=True), + sa.ForeignKey("chat_sessions.id", ondelete="CASCADE"), + nullable=False, + unique=True, + ), + sa.Column("rating", sa.Integer(), nullable=False), + sa.Column("comment", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.CheckConstraint("rating >= 1 AND rating <= 5", name="ck_session_feedback_rating_range"), + ) + + +def downgrade() -> None: + op.drop_table("session_feedback") diff --git a/dashboard/prisma/schema.prisma b/dashboard/prisma/schema.prisma index ab7a68e..72534f3 100644 --- a/dashboard/prisma/schema.prisma +++ b/dashboard/prisma/schema.prisma @@ -196,15 +196,31 @@ model ChatSession { llmContext Json @default("{}") @map("llm_context") availableFunctionNames Json @default("[]") @map("available_function_names") locale String @default("en") @db.VarChar(16) + escalatedAt DateTime? @map("escalated_at") @db.Timestamptz(6) + escalationReason String? @map("escalation_reason") + resolvedBy String? @map("resolved_by") @db.VarChar(20) app App @relation(fields: [appId], references: [id], onDelete: Cascade) messages Message[] llmUsageEvents LlmUsageEvent[] + feedback SessionFeedback? @@index([appId], map: "ix_chat_sessions_app_id") @@map("chat_sessions") } +model SessionFeedback { + id String @id @db.Uuid + createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(6) + sessionId String @unique @map("session_id") @db.Uuid + rating Int + comment String? + + session ChatSession @relation(fields: [sessionId], references: [id], onDelete: Cascade) + + @@map("session_feedback") +} + model LlmUsageEvent { id String @id @db.Uuid createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(6) diff --git a/dashboard/src/api/client.ts b/dashboard/src/api/client.ts index 1483a3d..aca820e 100644 --- a/dashboard/src/api/client.ts +++ b/dashboard/src/api/client.ts @@ -44,7 +44,16 @@ function shouldAttachBearerToken(path: string): boolean { } function toRequestUrl(path: string): string { - return shouldUseSameOriginAuthRoute(path) ? path : `${BASE}${path}`; + // The `dashboard` and `api` deploy targets are the same Next.js app/image in + // every topology this repo supports (docker-compose, staging, prod) — they + // just run as separate replicas on different ports/domains for routing + // convenience. A same-origin request always reaches an instance with the + // exact same /v1/* handlers and database, so route everything same-origin + // instead of a build-time NEXT_PUBLIC_API_BASE_URL, which is baked once in + // CI and would otherwise point every non-cookie-bound deployment (e.g. + // staging) at whatever origin happened to be configured for that build. + void BASE; + return path; } export async function logout(): Promise { diff --git a/dashboard/src/app/v1/apps/[appId]/analytics/route.ts b/dashboard/src/app/v1/apps/[appId]/analytics/route.ts new file mode 100644 index 0000000..57ab9a2 --- /dev/null +++ b/dashboard/src/app/v1/apps/[appId]/analytics/route.ts @@ -0,0 +1,98 @@ +import { Prisma } from "@prisma/client"; +import { NextRequest, NextResponse } from "next/server"; + +import { getDeveloperFromRequest } from "@/lib/server/auth"; +import { getOwnedAppOrNull } from "@/lib/server/apps"; +import { detail } from "@/lib/server/http"; +import { coerceDbNumber } from "@/lib/server/numbers"; +import { prisma } from "@/lib/server/prisma"; + +export const dynamic = "force-dynamic"; + +type DailyRow = { + day: Date; + total: number | string | bigint; + resolved: number | string | bigint; + escalated: number | string | bigint; +}; + +export async function GET(request: NextRequest, context: { params: Promise<{ appId: string }> }) { + const developer = await getDeveloperFromRequest(request); + if (!developer) return detail(401, "Invalid token"); + + const { appId } = await context.params; + const app = await getOwnedAppOrNull(appId, developer); + if (!app) return detail(404, "App not found"); + + const fromParam = request.nextUrl.searchParams.get("from"); + const toParam = request.nextUrl.searchParams.get("to"); + const to = toParam ? new Date(toParam) : new Date(); + const from = fromParam ? new Date(fromParam) : new Date(to.getTime() - 30 * 24 * 60 * 60 * 1000); + if (Number.isNaN(from.valueOf()) || Number.isNaN(to.valueOf())) { + return detail(400, "Invalid from/to date"); + } + + const [totalSessions, resolvedSessions, escalatedByStatus, escalatedByResolver, feedbackAgg, dailyRows] = + await Promise.all([ + prisma.chatSession.count({ + where: { appId: app.id, createdAt: { gte: from, lte: to } }, + }), + prisma.chatSession.count({ + where: { appId: app.id, createdAt: { gte: from, lte: to }, resolvedBy: "ai" }, + }), + prisma.chatSession.count({ + where: { appId: app.id, createdAt: { gte: from, lte: to }, status: "escalated" }, + }), + prisma.chatSession.count({ + where: { appId: app.id, createdAt: { gte: from, lte: to }, resolvedBy: "human" }, + }), + prisma.sessionFeedback.aggregate({ + where: { session: { appId: app.id, createdAt: { gte: from, lte: to } } }, + _avg: { rating: true }, + _count: { rating: true }, + }), + prisma.$queryRaw(Prisma.sql` + SELECT + date_trunc('day', created_at) AS day, + COUNT(*)::bigint AS total, + COUNT(*) FILTER (WHERE resolved_by = 'ai')::bigint AS resolved, + COUNT(*) FILTER (WHERE status = 'escalated' OR resolved_by = 'human')::bigint AS escalated + FROM chat_sessions + WHERE app_id = ${app.id}::uuid + AND created_at >= ${from} + AND created_at <= ${to} + GROUP BY 1 + ORDER BY 1 + `), + ]); + + const ratingDistribution = await prisma.sessionFeedback.groupBy({ + by: ["rating"], + where: { session: { appId: app.id, createdAt: { gte: from, lte: to } } }, + _count: { rating: true }, + }); + + const escalatedSessions = Math.max(escalatedByStatus, escalatedByResolver); + + return NextResponse.json({ + from: from.toISOString(), + to: to.toISOString(), + total_sessions: totalSessions, + resolved_sessions: resolvedSessions, + escalated_sessions: escalatedSessions, + abandoned_sessions: Math.max(0, totalSessions - resolvedSessions - escalatedSessions), + resolution_rate: totalSessions > 0 ? resolvedSessions / totalSessions : 0, + escalation_rate: totalSessions > 0 ? escalatedSessions / totalSessions : 0, + avg_csat: feedbackAgg._avg.rating ?? null, + csat_response_count: feedbackAgg._count.rating, + csat_distribution: ratingDistribution + .map((row) => ({ rating: row.rating, count: row._count.rating })) + .sort((a, b) => a.rating - b.rating), + daily: dailyRows.map((row) => ({ + date: row.day.toISOString().slice(0, 10), + total: coerceDbNumber(row.total), + resolved: coerceDbNumber(row.resolved), + escalated: coerceDbNumber(row.escalated), + })), + }); +} diff --git a/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/escalate/route.ts b/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/escalate/route.ts new file mode 100644 index 0000000..5a9c727 --- /dev/null +++ b/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/escalate/route.ts @@ -0,0 +1,43 @@ +import { NextRequest, NextResponse } from "next/server"; + +import { getDeveloperFromRequest } from "@/lib/server/auth"; +import { getOwnedAppOrNull } from "@/lib/server/apps"; +import { detail } from "@/lib/server/http"; +import { postSessionEscalated } from "@/lib/server/agent-service"; +import { prisma } from "@/lib/server/prisma"; +import { sessionOut } from "@/lib/server/serializers"; + +export const dynamic = "force-dynamic"; + +export async function POST( + request: NextRequest, + context: { params: Promise<{ appId: string; sessionId: string }> }, +) { + const developer = await getDeveloperFromRequest(request); + if (!developer) return detail(401, "Invalid token"); + + const { appId, sessionId } = await context.params; + const app = await getOwnedAppOrNull(appId, developer); + if (!app) return detail(404, "App not found"); + + const session = await prisma.chatSession.findUnique({ where: { id: sessionId } }); + if (!session || session.appId !== app.id) return detail(404, "Session not found"); + if (session.status === "closed") return detail(409, "Session is already closed"); + + const reason = `Manually taken over by ${developer.name}`; + const updated = await prisma.chatSession.update({ + where: { id: sessionId }, + data: { + status: "escalated", + escalatedAt: new Date(), + escalationReason: reason, + }, + }); + + // Best-effort: notify any connected client so it can show the escalated + // state and suppress a stale CSAT prompt. Don't fail the takeover if the + // agent is unreachable. + await postSessionEscalated(sessionId, reason).catch(() => {}); + + return NextResponse.json(sessionOut(updated)); +} diff --git a/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/reply/route.ts b/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/reply/route.ts new file mode 100644 index 0000000..27b6c5e --- /dev/null +++ b/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/reply/route.ts @@ -0,0 +1,34 @@ +import { NextRequest, NextResponse } from "next/server"; + +import { getDeveloperFromRequest } from "@/lib/server/auth"; +import { getOwnedAppOrNull } from "@/lib/server/apps"; +import { detail } from "@/lib/server/http"; +import { prisma } from "@/lib/server/prisma"; +import { postHumanMessage } from "@/lib/server/agent-service"; + +export const dynamic = "force-dynamic"; + +export async function POST( + request: NextRequest, + context: { params: Promise<{ appId: string; sessionId: string }> }, +) { + const developer = await getDeveloperFromRequest(request); + if (!developer) return detail(401, "Invalid token"); + + const { appId, sessionId } = await context.params; + const app = await getOwnedAppOrNull(appId, developer); + if (!app) return detail(404, "App not found"); + + const session = await prisma.chatSession.findUnique({ where: { id: sessionId } }); + if (!session || session.appId !== app.id) return detail(404, "Session not found"); + + const body = await request.json().catch(() => null); + const text = typeof body?.text === "string" ? body.text.trim() : ""; + if (!text) return detail(400, "text is required"); + + // The agent service owns the Message write and pushes it onto the session's + // live SSE stream — do not also write it here, or the message would be duplicated. + const message = await postHumanMessage(sessionId, text); + + return NextResponse.json(message); +} diff --git a/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/resolve/route.ts b/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/resolve/route.ts new file mode 100644 index 0000000..fd487de --- /dev/null +++ b/dashboard/src/app/v1/apps/[appId]/sessions/[sessionId]/resolve/route.ts @@ -0,0 +1,36 @@ +import { NextRequest, NextResponse } from "next/server"; + +import { getDeveloperFromRequest } from "@/lib/server/auth"; +import { getOwnedAppOrNull } from "@/lib/server/apps"; +import { detail } from "@/lib/server/http"; +import { postFeedbackRequested } from "@/lib/server/agent-service"; +import { prisma } from "@/lib/server/prisma"; +import { sessionOut } from "@/lib/server/serializers"; + +export const dynamic = "force-dynamic"; + +export async function POST( + request: NextRequest, + context: { params: Promise<{ appId: string; sessionId: string }> }, +) { + const developer = await getDeveloperFromRequest(request); + if (!developer) return detail(401, "Invalid token"); + + const { appId, sessionId } = await context.params; + const app = await getOwnedAppOrNull(appId, developer); + if (!app) return detail(404, "App not found"); + + const session = await prisma.chatSession.findUnique({ where: { id: sessionId } }); + if (!session || session.appId !== app.id) return detail(404, "Session not found"); + + const updated = await prisma.chatSession.update({ + where: { id: sessionId }, + data: { status: "closed", resolvedBy: "human" }, + }); + + // Best-effort: prompt the user for CSAT now that a human has resolved the + // conversation. Don't fail the resolve action if the agent is unreachable. + await postFeedbackRequested(sessionId).catch(() => {}); + + return NextResponse.json(sessionOut(updated)); +} diff --git a/dashboard/src/components/AppSidebar.tsx b/dashboard/src/components/AppSidebar.tsx index fcda323..58594eb 100644 --- a/dashboard/src/components/AppSidebar.tsx +++ b/dashboard/src/components/AppSidebar.tsx @@ -26,12 +26,14 @@ const CUSTOMIZATION_ITEMS: NavItem[] = [ ]; const CUSTOMIZATION_PARENT_ITEM: NavItem = { label: "Customization", slug: "chat-theme" }; const SESSIONS_ITEM: NavItem = { label: "Sessions", slug: "sessions" }; +const ANALYTICS_ITEM: NavItem = { label: "Analytics", slug: "analytics" }; const AUDIT_ITEM: NavItem = { label: "Audit Log", slug: "audit" }; const ROUTE_LABEL_ITEMS: NavItem[] = [ ...AGENT_CHILD_ITEMS, API_KEYS_ITEM, ...CUSTOMIZATION_ITEMS, SESSIONS_ITEM, + ANALYTICS_ITEM, AUDIT_ITEM, ]; const AGENT_SECTION_SLUGS = AGENT_CHILD_ITEMS.map((item) => item.slug); @@ -218,6 +220,7 @@ export default function AppSidebar({ variant = "desktop" }: AppSidebarProps) { )} {renderNavButton(SESSIONS_ITEM)} + {renderNavButton(ANALYTICS_ITEM)} {renderNavButton(AUDIT_ITEM)} @@ -280,6 +283,7 @@ export default function AppSidebar({ variant = "desktop" }: AppSidebarProps) { )} {renderNavButton(SESSIONS_ITEM)} + {renderNavButton(ANALYTICS_ITEM)} {renderNavButton(AUDIT_ITEM)} diff --git a/dashboard/src/components/ui/AppNav.tsx b/dashboard/src/components/ui/AppNav.tsx index 33d54ab..266d762 100644 --- a/dashboard/src/components/ui/AppNav.tsx +++ b/dashboard/src/components/ui/AppNav.tsx @@ -18,6 +18,7 @@ const TABS = [ { label: "Chat Theme", slug: "chat-theme" }, { label: "Localization", slug: "languages" }, { label: "Sessions", slug: "sessions" }, + { label: "Analytics", slug: "analytics" }, { label: "Audit Log", slug: "audit" }, ]; diff --git a/dashboard/src/dashboard-app.tsx b/dashboard/src/dashboard-app.tsx index 111f53f..63cb7f0 100644 --- a/dashboard/src/dashboard-app.tsx +++ b/dashboard/src/dashboard-app.tsx @@ -14,6 +14,7 @@ import AppKnowledgeBases from "./dashboard_pages/AppKnowledgeBases"; import OrganizationAdmin from "./dashboard_pages/OrganizationAdmin"; import Playbooks from "./dashboard_pages/Playbooks"; import Sessions from "./dashboard_pages/Sessions"; +import Analytics from "./dashboard_pages/Analytics"; import ChatTheme from "./dashboard_pages/ChatTheme"; export function RootRedirect() { @@ -37,6 +38,7 @@ export function DashboardApp() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/dashboard/src/dashboard_pages/Analytics.tsx b/dashboard/src/dashboard_pages/Analytics.tsx new file mode 100644 index 0000000..1e1d43b --- /dev/null +++ b/dashboard/src/dashboard_pages/Analytics.tsx @@ -0,0 +1,243 @@ +import { useEffect, useMemo, useState } from "react"; +import { useParams } from "react-router-dom"; + +import { api } from "../api/client"; +import { PageSpinner } from "../components/ui"; +import { PageHeader } from "../components/layout/PageHeader"; + +interface DailyPoint { + date: string; + total: number; + resolved: number; + escalated: number; +} + +interface AnalyticsSummary { + total_sessions: number; + resolved_sessions: number; + escalated_sessions: number; + abandoned_sessions: number; + resolution_rate: number; + escalation_rate: number; + avg_csat: number | null; + csat_response_count: number; + csat_distribution: Array<{ rating: number; count: number }>; + daily: DailyPoint[]; +} + +function formatPercent(value: number): string { + return `${Math.round(value * 100)}%`; +} + +function StatTile({ label, value, sublabel }: { label: string; value: string; sublabel?: string }) { + return ( +
+

{label}

+

{value}

+ {sublabel &&

{sublabel}

} +
+ ); +} + +function DailyChart({ daily }: { daily: DailyPoint[] }) { + const [hoverIndex, setHoverIndex] = useState(null); + + const width = 640; + const height = 220; + const padding = { top: 12, right: 12, bottom: 24, left: 12 }; + const plotWidth = width - padding.left - padding.right; + const plotHeight = height - padding.top - padding.bottom; + + const maxTotal = Math.max(1, ...daily.map((d) => d.total)); + const stepX = daily.length > 1 ? plotWidth / (daily.length - 1) : 0; + + const points = (key: "resolved" | "escalated") => + daily + .map((d, i) => { + const x = padding.left + i * stepX; + const y = padding.top + plotHeight - (d[key] / maxTotal) * plotHeight; + return `${x},${y}`; + }) + .join(" "); + + if (daily.length === 0) { + return

No session activity in this range.

; + } + + const hovered = hoverIndex !== null ? daily[hoverIndex] : null; + + return ( +
+ setHoverIndex(null)} + onMouseMove={(e) => { + const rect = e.currentTarget.getBoundingClientRect(); + const relX = ((e.clientX - rect.left) / rect.width) * width - padding.left; + const idx = stepX > 0 ? Math.round(relX / stepX) : 0; + setHoverIndex(Math.min(daily.length - 1, Math.max(0, idx))); + }} + > + + + + {hoverIndex !== null && ( + + )} + + {hovered && ( +
+

{hovered.date}

+

resolved: {hovered.resolved}

+

escalated: {hovered.escalated}

+
+ )} +
+ + Resolved (AI) + + + Escalated + +
+
+ ); +} + +export default function Analytics() { + const { appId } = useParams<{ appId: string }>(); + const [data, setData] = useState(null); + const [isLoading, setIsLoading] = useState(true); + const [error, setError] = useState(null); + const [rangeDays, setRangeDays] = useState(30); + + useEffect(() => { + setIsLoading(true); + setError(null); + const to = new Date(); + const from = new Date(to.getTime() - rangeDays * 24 * 60 * 60 * 1000); + api( + `/v1/apps/${appId}/analytics?from=${from.toISOString()}&to=${to.toISOString()}` + ) + .then(setData) + .catch(() => setError("Failed to load analytics")) + .finally(() => setIsLoading(false)); + }, [appId, rangeDays]); + + const maxDistCount = useMemo( + () => Math.max(1, ...(data?.csat_distribution.map((row) => row.count) ?? [])), + [data] + ); + + if (isLoading) return ; + + return ( +
+ + +
+ +
+ + {error &&

{error}

} + + {data && ( + <> +
+ + + + +
+ +
+
+

Daily Sessions

+ +
+ +
+

CSAT Distribution

+ {data.csat_distribution.length === 0 ? ( +

No feedback submitted yet.

+ ) : ( +
+ {[1, 2, 3, 4, 5].map((rating) => { + const row = data.csat_distribution.find((r) => r.rating === rating); + const count = row?.count ?? 0; + return ( +
+ {rating}★ +
+
+
+ {count} +
+ ); + })} +
+ )} +
+
+ + )} +
+ ); +} diff --git a/dashboard/src/dashboard_pages/Sessions.tsx b/dashboard/src/dashboard_pages/Sessions.tsx index 9f05b01..2fd9543 100644 --- a/dashboard/src/dashboard_pages/Sessions.tsx +++ b/dashboard/src/dashboard_pages/Sessions.tsx @@ -14,6 +14,7 @@ interface Session { client_context?: Record; llm_context?: Record; cost_summary?: SessionCostSummary | null; + escalation_reason?: string | null; } interface Message { @@ -71,10 +72,11 @@ type KbSearchResult = { error: string | null; }; -function statusVariant(status: string): "active" | "expired" | "closed" | "default" { +function statusVariant(status: string): "active" | "expired" | "closed" | "danger" | "default" { if (status === "active") return "active"; if (status === "expired") return "expired"; if (status === "closed") return "closed"; + if (status === "escalated") return "danger"; return "default"; } @@ -285,6 +287,11 @@ export default function Sessions() { const [selectedCostSummary, setSelectedCostSummary] = useState(null); const [costsLoading, setCostsLoading] = useState(false); const [costsError, setCostsError] = useState(null); + const [replyText, setReplyText] = useState(""); + const [isSendingReply, setIsSendingReply] = useState(false); + const [replyError, setReplyError] = useState(null); + const [isResolving, setIsResolving] = useState(false); + const [isTakingOver, setIsTakingOver] = useState(false); const abortRef = useRef(null); const pollRef = useRef | null>(null); @@ -331,6 +338,8 @@ export default function Sessions() { setSelectedCostSummary(sessions.find((entry) => entry.id === sessionId)?.cost_summary ?? null); setCostsError(null); setCostsLoading(true); + setReplyText(""); + setReplyError(null); if (pollRef.current) { clearInterval(pollRef.current); @@ -353,7 +362,7 @@ export default function Sessions() { setCostsLoading(false); const session = sessions.find((entry) => entry.id === sessionId); - if (session?.status === "active") { + if (session?.status === "active" || session?.status === "escalated") { pollRef.current = setInterval(async () => { try { const updated = await api(`/v1/apps/${appId}/sessions/${sessionId}/messages`); @@ -382,6 +391,68 @@ export default function Sessions() { }; }, []); + async function sendReply() { + const sessionId = selectedId; + const text = replyText.trim(); + if (!sessionId || !text) return; + setIsSendingReply(true); + setReplyError(null); + try { + const message = await api(`/v1/apps/${appId}/sessions/${sessionId}/reply`, { + method: "POST", + body: JSON.stringify({ text }), + }); + setMessages((prev) => [...prev, message]); + setReplyText(""); + } catch (err: unknown) { + setReplyError(err instanceof ApiError ? err.detail : "Failed to send reply"); + } finally { + setIsSendingReply(false); + } + } + + async function takeOverSession() { + const sessionId = selectedId; + if (!sessionId) return; + setIsTakingOver(true); + setReplyError(null); + try { + const updated = await api(`/v1/apps/${appId}/sessions/${sessionId}/escalate`, { + method: "POST", + }); + setSessions((prev) => prev.map((entry) => (entry.id === sessionId ? updated : entry))); + if (pollRef.current) clearInterval(pollRef.current); + pollRef.current = setInterval(async () => { + try { + const updatedMessages = await api(`/v1/apps/${appId}/sessions/${sessionId}/messages`); + setMessages(updatedMessages); + } catch { + // Polling should not disrupt the current view. + } + }, 10000); + } catch (err: unknown) { + setReplyError(err instanceof ApiError ? err.detail : "Failed to take over session"); + } finally { + setIsTakingOver(false); + } + } + + async function resolveSession() { + const sessionId = selectedId; + if (!sessionId) return; + setIsResolving(true); + try { + const updated = await api(`/v1/apps/${appId}/sessions/${sessionId}/resolve`, { + method: "POST", + }); + setSessions((prev) => prev.map((entry) => (entry.id === sessionId ? updated : entry))); + } catch (err: unknown) { + setReplyError(err instanceof ApiError ? err.detail : "Failed to resolve session"); + } finally { + setIsResolving(false); + } + } + const selectedSession = sessions.find((session) => session.id === selectedId); const toolResultByCallId = useMemo(() => { const map = new Map(); @@ -569,6 +640,37 @@ export default function Sessions() {
)} {selectedSession.llm_context && } + {selectedSession.status === "active" && ( +
+

Join this conversation as a human agent at any time.

+ +
+ )} + {selectedSession.status === "escalated" && ( +
+
+

Escalated to human

+ {selectedSession.escalation_reason && ( +

{selectedSession.escalation_reason}

+ )} +
+ +
+ )}
@@ -628,6 +730,29 @@ export default function Sessions() { }) )}
+ + {selectedSession.status === "escalated" && ( +
+ {replyError &&

{replyError}

} +
+