Add human handoff, CSAT capture, and analytics dashboard

.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

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()

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

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",

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,21 +11,26 @@
 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):
     __tablename__ = "chat_sessions"
 
     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)
     )
     client_context: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict)
     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)

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")

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"}

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

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