Save text records to db

engine/core/session.py

@@ -2,12 +2,19 @@
 
 import uuid
 import json
+import time
 from enum import Enum
 from typing import Optional, Dict, Any
 from loguru import logger
 
+from app.backend_client import (
+    create_history_call_record,
+    add_history_transcript,
+    finalize_history_call_record,
+)
 from core.transports import BaseTransport
 from core.duplex_pipeline import DuplexPipeline
+from core.conversation import ConversationTurn
 from app.config import settings
 from models.ws_v1 import (
     parse_client_message,
@@ -67,6 +74,12 @@ class Session:
 
         # Track IDs
         self.current_track_id: Optional[str] = str(uuid.uuid4())
+        self._history_call_id: Optional[str] = None
+        self._history_turn_index: int = 0
+        self._history_call_started_mono: Optional[float] = None
+        self._history_finalized: bool = False
+
+        self.pipeline.conversation.on_turn_complete(self._on_turn_complete)
 
         logger.info(f"Session {self.id} created (duplex={self.use_duplex})")
 
@@ -206,8 +219,13 @@ class Session:
             await self._send_error("client", "Duplicate session.start", "protocol.order")
             return
 
+        metadata = message.metadata or {}
+
+        # Create history call record early so later turn callbacks can append transcripts.
+        await self._start_history_bridge(metadata)
+
         # Apply runtime service/prompt overrides from backend if provided
-        self.pipeline.apply_runtime_overrides(message.metadata)
+        self.pipeline.apply_runtime_overrides(metadata)
 
         # Start duplex pipeline
         if not self._pipeline_started:
@@ -241,6 +259,7 @@ class Session:
                 reason=stop_reason,
             )
         )
+        await self._finalize_history(status="connected")
         await self.transport.close()
 
     async def _send_error(self, sender: str, error_message: str, code: str) -> None:
@@ -270,5 +289,83 @@ class Session:
     async def cleanup(self) -> None:
         """Cleanup session resources."""
         logger.info(f"Session {self.id} cleaning up")
+        await self._finalize_history(status="connected")
         await self.pipeline.cleanup()
         await self.transport.close()
+
+    async def _start_history_bridge(self, metadata: Dict[str, Any]) -> None:
+        """Initialize backend history call record for this session."""
+        if self._history_call_id:
+            return
+
+        history_meta: Dict[str, Any] = {}
+        if isinstance(metadata.get("history"), dict):
+            history_meta = metadata["history"]
+
+        raw_user_id = history_meta.get("userId", metadata.get("userId", settings.history_default_user_id))
+        try:
+            user_id = int(raw_user_id)
+        except (TypeError, ValueError):
+            user_id = settings.history_default_user_id
+
+        assistant_id = history_meta.get("assistantId", metadata.get("assistantId"))
+        source = str(history_meta.get("source", metadata.get("source", "debug")))
+
+        call_id = await create_history_call_record(
+            user_id=user_id,
+            assistant_id=str(assistant_id) if assistant_id else None,
+            source=source,
+        )
+        if not call_id:
+            return
+
+        self._history_call_id = call_id
+        self._history_call_started_mono = time.monotonic()
+        self._history_turn_index = 0
+        self._history_finalized = False
+        logger.info(f"Session {self.id} history bridge enabled (call_id={call_id}, source={source})")
+
+    async def _on_turn_complete(self, turn: ConversationTurn) -> None:
+        """Persist completed turns to backend call transcripts."""
+        if not self._history_call_id:
+            return
+        if not turn.text or not turn.text.strip():
+            return
+
+        role = (turn.role or "").lower()
+        speaker = "human" if role == "user" else "ai"
+
+        end_ms = 0
+        if self._history_call_started_mono is not None:
+            end_ms = max(0, int((time.monotonic() - self._history_call_started_mono) * 1000))
+        estimated_duration_ms = max(300, min(12000, len(turn.text.strip()) * 80))
+        start_ms = max(0, end_ms - estimated_duration_ms)
+
+        turn_index = self._history_turn_index
+        await add_history_transcript(
+            call_id=self._history_call_id,
+            turn_index=turn_index,
+            speaker=speaker,
+            content=turn.text.strip(),
+            start_ms=start_ms,
+            end_ms=end_ms,
+            duration_ms=max(1, end_ms - start_ms),
+        )
+        self._history_turn_index += 1
+
+    async def _finalize_history(self, status: str) -> None:
+        """Finalize history call record once."""
+        if not self._history_call_id or self._history_finalized:
+            return
+
+        duration_seconds = 0
+        if self._history_call_started_mono is not None:
+            duration_seconds = max(0, int(time.monotonic() - self._history_call_started_mono))
+
+        ok = await finalize_history_call_record(
+            call_id=self._history_call_id,
+            status=status,
+            duration_seconds=duration_seconds,
+        )
+        if ok:
+            self._history_finalized = True