apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-06-10 08:05:22 +02:00
Code Issues Projects Releases Packages Wiki Activity

chore: fix tracing for text chat mode

Browse source
This commit is contained in:
Abhishek Kumar 2026-05-21 12:30:56 +05:30
parent e23cce444f
commit 08a2435ba5
31 changed files with 1753 additions and 597 deletions
Download patch file Download diff file Expand all files Collapse all files

4
api/AGENTS.md
View file

@ -41,6 +41,10 @@ api/
- Telephony is a full subsystem under `services/telephony/`, with provider-specific packages under `services/telephony/providers/`
- Integrations extend through `services/integrations/`; package-specific rules should live in that subtree's own `AGENTS.md`
## Routes vs Service Layer
**Keep route handlers thin** — parse/validate the request, resolve auth and `organization_id`, delegate, shape the response. Domain logic (orchestration, business rules, external calls, computation) belongs in `services/`. Before adding logic to a handler, find its home: extend an existing `services/<domain>/` module that owns the concern (see *Where to Find Things*) before adding a focused new module; never a catch-all. Keep DB access in `db/` clients — routes call services, services call DB clients. Litmus test: if `tasks/`, `mcp_server/`, or another route could reuse it, it must live in `services/` to be importable.
## Database Migrations
```bash

1
api/alembic/versions/19d2a4b6c8ef_rename_integrations_organisation_id.py
View file

@ -2,6 +2,7 @@
Revision ID: 19d2a4b6c8ef
Revises: 0a1b2c3d4e5f
Create Date: 2026-05-19 00:00:00.000000
"""

4
api/alembic/versions/2f638891cbb6_add_workflow_run_text_sessions.py
View file

@ -1,7 +1,7 @@
"""add workflow_run_text_sessions
Revision ID: 2f638891cbb6
Revises: 4c1f1e3e8ef2
Revises: 19d2a4b6c8ef
Create Date: 2026-05-18 12:58:58.573381
"""
@ -13,7 +13,7 @@ from alembic import op
# revision identifiers, used by Alembic.
revision: str = "2f638891cbb6"
down_revision: Union[str, None] = "4c1f1e3e8ef2"
down_revision: Union[str, None] = "19d2a4b6c8ef"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

1
api/db/organization_usage_client.py
View file

@ -354,6 +354,7 @@ class OrganizationUsageClient(BaseDBClient):
"caller_number": caller_number,
"called_number": called_number,
"call_type": run.call_type,
"mode": run.mode,
"disposition": disposition,
"initial_context": run.initial_context,
"gathered_context": run.gathered_context,

1
api/routes/organization_usage.py
View file

@ -57,6 +57,7 @@ class WorkflowRunUsageResponse(BaseModel):
caller_number: Optional[str] = None
called_number: Optional[str] = None
call_type: Optional[str] = None
mode: Optional[str] = None
disposition: Optional[str] = None
initial_context: Optional[Dict[str, Any]] = None
gathered_context: Optional[Dict[str, Any]] = None

319
api/routes/workflow_text_chat.py
View file

@ -1,27 +1,32 @@
from datetime import UTC, datetime
from datetime import datetime
from typing import Any, Dict
from uuid import uuid4
from fastapi import APIRouter, Depends, HTTPException
from pipecat.utils.run_context import set_current_run_id
from pydantic import BaseModel, Field
from api.db import db_client
from api.db.models import UserModel, WorkflowRunTextSessionModel
from api.db.workflow_run_text_session_client import (
WorkflowRunTextSessionRevisionConflictError,
)
from api.enums import WorkflowRunMode
from api.services.auth.depends import get_user
from api.services.workflow.text_chat_runner import (
execute_text_chat_pending_turn,
merge_text_chat_usage_info,
from api.services.workflow.text_chat_session_service import (
TextChatPendingTurnLostError,
TextChatSessionExecutionError,
TextChatSessionRevisionConflictError,
TextChatTurnNotFoundError,
append_text_chat_user_message,
default_text_chat_checkpoint,
default_text_chat_session_data,
execute_pending_text_chat_turn,
initialize_text_chat_session,
normalize_text_chat_checkpoint,
normalize_text_chat_session_data,
rewind_text_chat_session_state,
)
router = APIRouter(prefix="/workflow", tags=["workflow-text-chat"])
TEXT_CHAT_SESSION_VERSION = 1
TEXT_CHAT_CHECKPOINT_VERSION = 1
class CreateTextChatSessionRequest(BaseModel):
name: str | None = None
@ -56,57 +61,6 @@ class WorkflowRunTextSessionResponse(BaseModel):
updated_at: datetime | None = None
def _default_session_data() -> Dict[str, Any]:
return {
"version": TEXT_CHAT_SESSION_VERSION,
"status": "idle",
"cursor_turn_id": None,
"turns": [],
"discarded_future": [],
"simulator": {
"enabled": False,
"config": {},
},
}
def _default_checkpoint() -> Dict[str, Any]:
return {
"version": TEXT_CHAT_CHECKPOINT_VERSION,
"anchor_turn_id": None,
"current_node_id": None,
"messages": [],
"gathered_context": {},
"tool_state": {},
}
def _normalize_session_data(session_data: Dict[str, Any] | None) -> Dict[str, Any]:
normalized = {
**_default_session_data(),
**(session_data or {}),
}
normalized["turns"] = list(normalized.get("turns") or [])
normalized["discarded_future"] = list(normalized.get("discarded_future") or [])
simulator = normalized.get("simulator") or {}
normalized["simulator"] = {
"enabled": bool(simulator.get("enabled", False)),
"config": dict(simulator.get("config") or {}),
}
return normalized
def _normalize_checkpoint(checkpoint: Dict[str, Any] | None) -> Dict[str, Any]:
normalized = {
**_default_checkpoint(),
**(checkpoint or {}),
}
normalized["messages"] = list(normalized.get("messages") or [])
normalized["gathered_context"] = dict(normalized.get("gathered_context") or {})
normalized["tool_state"] = dict(normalized.get("tool_state") or {})
return normalized
def _get_state_value(state: Any) -> str:
return state.value if hasattr(state, "value") else str(state)
@ -126,81 +80,14 @@ def _build_response(
initial_context=workflow_run.initial_context,
gathered_context=workflow_run.gathered_context,
annotations=workflow_run.annotations,
session_data=_normalize_session_data(text_session.session_data),
checkpoint=_normalize_checkpoint(text_session.checkpoint),
session_data=normalize_text_chat_session_data(text_session.session_data),
checkpoint=normalize_text_chat_checkpoint(text_session.checkpoint),
created_at=text_session.created_at,
updated_at=text_session.updated_at,
)
def _validate_turn_cursor(
session_data: Dict[str, Any], cursor_turn_id: str | None
) -> None:
if cursor_turn_id is None:
return
if not any(turn.get("id") == cursor_turn_id for turn in session_data["turns"]):
raise HTTPException(
status_code=404, detail="Turn not found in text chat session"
)
def _truncate_future_turns(
session_data: Dict[str, Any],
) -> tuple[list[Dict[str, Any]], list[Dict[str, Any]]]:
cursor_turn_id = session_data.get("cursor_turn_id")
turns = list(session_data.get("turns") or [])
discarded_future = list(session_data.get("discarded_future") or [])
if cursor_turn_id is None:
return turns, discarded_future
for index, turn in enumerate(turns):
if turn.get("id") == cursor_turn_id:
active_turns = turns[: index + 1]
future_turns = turns[index + 1 :]
if future_turns:
discarded_future.append(
{
"rewound_from_turn_id": cursor_turn_id,
"discarded_at": datetime.now(UTC).isoformat(),
"turns": future_turns,
}
)
return active_turns, discarded_future
raise HTTPException(status_code=404, detail="Turn not found in text chat session")
def _latest_completed_turn_id(turns: list[Dict[str, Any]]) -> str | None:
for turn in reversed(turns):
if turn.get("status") == "completed":
return turn.get("id")
return None
def _build_pending_turn(*, user_text: str | None) -> Dict[str, Any]:
now = datetime.now(UTC).isoformat()
return {
"id": f"turn_{uuid4().hex[:12]}",
"status": "pending",
"created_at": now,
"user_message": (
{
"text": user_text,
"created_at": now,
}
if user_text is not None
else None
),
"assistant_message": None,
"events": [],
"usage": {},
}
def _revision_conflict_detail(
e: WorkflowRunTextSessionRevisionConflictError,
) -> dict[str, Any]:
def _revision_conflict_detail(e: Any) -> dict[str, Any]:
return {
"message": "Text chat session revision conflict",
"expected_revision": e.expected_revision,
@ -213,6 +100,7 @@ async def _load_text_session_or_404(
run_id: int,
user: UserModel,
) -> WorkflowRunTextSessionModel:
set_current_run_id(run_id)
if user.selected_organization_id is None:
raise HTTPException(status_code=403, detail="Organization context is required")
text_session = await db_client.get_workflow_run_text_session(
@ -229,96 +117,26 @@ async def _load_text_session_or_404(
return text_session
async def _execute_pending_turn_and_build_response(
async def _execute_pending_turn_response(
*,
workflow_id: int,
run_id: int,
text_session: WorkflowRunTextSessionModel,
user: UserModel,
) -> WorkflowRunTextSessionResponse:
try:
execution = await execute_text_chat_pending_turn(
workflow_run_id=run_id,
updated_text_session = await execute_pending_text_chat_turn(
workflow_id=workflow_id,
session_data=_normalize_session_data(text_session.session_data),
checkpoint=_normalize_checkpoint(text_session.checkpoint),
run_id=run_id,
text_session=text_session,
)
except Exception as e:
failed_session_data = _normalize_session_data(text_session.session_data)
failed_turns = list(failed_session_data.get("turns") or [])
if failed_turns and failed_turns[-1].get("status") == "pending":
failed_turns[-1]["status"] = "failed"
failed_turns[-1]["events"] = [
*(failed_turns[-1].get("events") or []),
{
"type": "execution_error",
"created_at": datetime.now(UTC).isoformat(),
"payload": {"message": str(e)},
},
]
failed_session_data["turns"] = failed_turns
failed_session_data["status"] = "error"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=failed_session_data,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError:
pass
raise HTTPException(
status_code=500, detail="Failed to execute text chat assistant turn"
)
completed_session_data = _normalize_session_data(text_session.session_data)
completed_turns = list(completed_session_data.get("turns") or [])
if not completed_turns or completed_turns[-1].get("status") != "pending":
raise HTTPException(
status_code=500,
detail="Text chat session lost its pending turn before completion",
)
completed_turns[-1]["status"] = "completed"
completed_turns[-1]["assistant_message"] = (
{
"text": execution.assistant_text,
"created_at": execution.assistant_created_at,
}
if execution.assistant_text
else None
)
completed_turns[-1]["events"] = execution.events
completed_turns[-1]["usage"] = execution.usage
completed_turns[-1]["checkpoint_after_turn"] = execution.checkpoint
completed_session_data["turns"] = completed_turns
completed_session_data["status"] = "idle"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=completed_session_data,
checkpoint=execution.checkpoint,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
except TextChatSessionRevisionConflictError as e:
raise HTTPException(status_code=409, detail=_revision_conflict_detail(e))
except TextChatPendingTurnLostError as e:
raise HTTPException(status_code=500, detail=str(e))
except TextChatSessionExecutionError as e:
raise HTTPException(status_code=500, detail=str(e))
existing_usage_info = text_session.workflow_run.usage_info or {}
merged_usage_info = merge_text_chat_usage_info(
existing_usage_info,
execution.usage,
)
await db_client.update_workflow_run(
run_id,
initial_context=execution.initial_context,
usage_info=merged_usage_info,
gathered_context=execution.gathered_context,
state=execution.state,
is_completed=execution.is_completed,
)
text_session = await _load_text_session_or_404(workflow_id, run_id, user)
return _build_response(text_session)
return _build_response(updated_text_session)
@router.post(
@ -343,6 +161,8 @@ async def create_text_chat_session(
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
set_current_run_id(workflow_run.id)
annotations = {
"tester": {
"source": "workflow_editor",
@ -358,32 +178,22 @@ async def create_text_chat_session(
text_session = await db_client.ensure_workflow_run_text_session(
workflow_run.id,
session_data=_default_session_data(),
checkpoint=_default_checkpoint(),
session_data=default_text_chat_session_data(),
checkpoint=default_text_chat_checkpoint(),
)
session_data = _normalize_session_data(text_session.session_data)
checkpoint = _normalize_checkpoint(text_session.checkpoint)
session_data["turns"] = [_build_pending_turn(user_text=None)]
session_data["status"] = "pending_assistant_turn"
checkpoint["anchor_turn_id"] = _latest_completed_turn_id(session_data["turns"])
try:
await db_client.update_workflow_run_text_session(
workflow_run.id,
session_data=session_data,
checkpoint=checkpoint,
expected_revision=text_session.revision,
text_session = await initialize_text_chat_session(
run_id=workflow_run.id,
text_session=text_session,
)
except WorkflowRunTextSessionRevisionConflictError as e:
except TextChatSessionRevisionConflictError as e:
raise HTTPException(status_code=409, detail=_revision_conflict_detail(e))
text_session = await _load_text_session_or_404(workflow_id, workflow_run.id, user)
return await _execute_pending_turn_and_build_response(
return await _execute_pending_turn_response(
workflow_id=workflow_id,
run_id=workflow_run.id,
text_session=text_session,
user=user,
)
@ -411,34 +221,20 @@ async def append_text_chat_message(
user: UserModel = Depends(get_user),
) -> WorkflowRunTextSessionResponse:
text_session = await _load_text_session_or_404(workflow_id, run_id, user)
session_data = _normalize_session_data(text_session.session_data)
checkpoint = _normalize_checkpoint(text_session.checkpoint)
active_turns, discarded_future = _truncate_future_turns(session_data)
active_turns.append(_build_pending_turn(user_text=request.text))
session_data["turns"] = active_turns
session_data["discarded_future"] = discarded_future
session_data["cursor_turn_id"] = None
session_data["status"] = "pending_assistant_turn"
checkpoint["anchor_turn_id"] = _latest_completed_turn_id(active_turns)
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
checkpoint=checkpoint,
text_session = await append_text_chat_user_message(
run_id=run_id,
text_session=text_session,
user_text=request.text,
expected_revision=request.expected_revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
except TextChatSessionRevisionConflictError as e:
raise HTTPException(status_code=409, detail=_revision_conflict_detail(e))
text_session = await _load_text_session_or_404(workflow_id, run_id, user)
return await _execute_pending_turn_and_build_response(
return await _execute_pending_turn_response(
workflow_id=workflow_id,
run_id=run_id,
text_session=text_session,
user=user,
)
@ -453,27 +249,16 @@ async def rewind_text_chat_session(
user: UserModel = Depends(get_user),
) -> WorkflowRunTextSessionResponse:
text_session = await _load_text_session_or_404(workflow_id, run_id, user)
session_data = _normalize_session_data(text_session.session_data)
_validate_turn_cursor(session_data, request.cursor_turn_id)
session_data["cursor_turn_id"] = request.cursor_turn_id
session_data["status"] = "rewound" if request.cursor_turn_id else "idle"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
text_session = await rewind_text_chat_session_state(
run_id=run_id,
text_session=text_session,
cursor_turn_id=request.cursor_turn_id,
expected_revision=request.expected_revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise HTTPException(
status_code=409,
detail={
"message": "Text chat session revision conflict",
"expected_revision": e.expected_revision,
"actual_revision": e.actual_revision,
},
)
except TextChatTurnNotFoundError as e:
raise HTTPException(status_code=404, detail=str(e))
except TextChatSessionRevisionConflictError as e:
raise HTTPException(status_code=409, detail=_revision_conflict_detail(e))
text_session = await _load_text_session_or_404(workflow_id, run_id, user)
return _build_response(text_session)

28
api/services/pipecat/in_memory_buffers.py
View file

@ -6,6 +6,10 @@ from typing import List, Optional
from loguru import logger
from api.services.pipecat.realtime_feedback_events import (
realtime_feedback_event_sort_key,
stamp_realtime_feedback_event,
)
from api.utils.transcript import generate_transcript_text as _generate_transcript_text
from pipecat.utils.enums import RealtimeFeedbackType
@ -98,16 +102,13 @@ class InMemoryLogsBuffer:
async def append(self, event: dict):
"""Append a feedback event to the buffer with timestamp and current node."""
# Add timestamp, turn tracking, and current node
timestamped_event = {
**event,
"timestamp": datetime.now(UTC).isoformat(),
"turn": self._turn_counter,
}
if self._current_node_id:
timestamped_event["node_id"] = self._current_node_id
if self._current_node_name:
timestamped_event["node_name"] = self._current_node_name
timestamped_event = stamp_realtime_feedback_event(
event,
timestamp=datetime.now(UTC).isoformat(),
turn=self._turn_counter,
node_id=self._current_node_id,
node_name=self._current_node_name,
)
self._events.append(timestamped_event)
logger.trace(
f"Appended event {event.get('type')} to logs buffer for workflow {self._workflow_run_id}"
@ -120,17 +121,12 @@ class InMemoryLogsBuffer:
f"Incremented turn counter to {self._turn_counter} for workflow {self._workflow_run_id}"
)
@staticmethod
def _event_sort_key(event: dict) -> str:
payload_ts = event.get("payload", {}).get("timestamp")
return payload_ts or event.get("timestamp", "")
def _sorted_events(self) -> List[dict]:
# Stable sort by the realtime (payload) timestamp when available, falling
# back to the buffer-append timestamp. Python's sort is stable, so events
# sharing a key retain their original insertion order — this keeps
# consecutive bot-text chunks of a single turn contiguous.
return sorted(self._events, key=self._event_sort_key)
return sorted(self._events, key=realtime_feedback_event_sort_key)
def get_events(self) -> List[dict]:
"""Get all events for final storage, ordered by realtime timestamp."""

29
api/services/pipecat/pipeline_builder.py
View file

@ -152,8 +152,30 @@ def build_realtime_pipeline(
return Pipeline(processors)
def create_pipeline_task(pipeline, workflow_run_id, audio_config: AudioConfig = None):
"""Create a pipeline task with appropriate parameters"""
def create_pipeline_task(
pipeline,
workflow_run_id,
audio_config: AudioConfig = None,
*,
conversation_parent_context=None,
conversation_type: str = "voice",
additional_span_attributes: dict | None = None,
):
"""Create a pipeline task with appropriate parameters.
Args:
pipeline: The pipeline to run.
workflow_run_id: Run id, used as the conversation id.
audio_config: Optional audio configuration.
conversation_parent_context: Optional OTEL context carrying a fixed
trace id. When provided, the conversation span attaches to that
trace instead of starting a new root trace (used by text chat to
stitch every per-turn pipeline into one trace).
conversation_type: ``conversation.type`` span attribute value.
additional_span_attributes: Extra attributes set on the conversation
span (e.g. ``langfuse.trace.name`` to name a stitched trace that
has no real root span).
"""
# Set up pipeline params with audio configuration if provided
pipeline_params = PipelineParams(
enable_metrics=True,
@ -178,6 +200,9 @@ def create_pipeline_task(pipeline, workflow_run_id, audio_config: AudioConfig =
enable_tracing=True,
enable_rtvi=False,
conversation_id=f"{workflow_run_id}",
conversation_parent_context=conversation_parent_context,
conversation_type=conversation_type,
additional_span_attributes=additional_span_attributes,
)
# Check if turn logging is enabled

159
api/services/pipecat/realtime_feedback_events.py Normal file
View file

@ -0,0 +1,159 @@
"""Shared helpers for building and ordering realtime feedback events."""
from typing import Any
from pipecat.utils.enums import RealtimeFeedbackType
def build_node_transition_event(
*,
node_id: str | None,
node_name: str | None,
previous_node_id: str | None,
previous_node_name: str | None,
allow_interrupt: bool = False,
) -> dict[str, Any]:
return {
"type": RealtimeFeedbackType.NODE_TRANSITION.value,
"payload": {
"node_id": node_id,
"node_name": node_name,
"previous_node_id": previous_node_id,
"previous_node_name": previous_node_name,
"allow_interrupt": allow_interrupt,
},
}
def build_user_transcription_event(
*,
text: str,
final: bool,
timestamp: str | None = None,
user_id: str | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"text": text,
"final": final,
}
if timestamp is not None:
payload["timestamp"] = timestamp
if user_id is not None:
payload["user_id"] = user_id
return {
"type": RealtimeFeedbackType.USER_TRANSCRIPTION.value,
"payload": payload,
}
def build_bot_text_event(
*,
text: str,
timestamp: str | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {"text": text}
if timestamp is not None:
payload["timestamp"] = timestamp
return {
"type": RealtimeFeedbackType.BOT_TEXT.value,
"payload": payload,
}
def build_function_call_start_event(
*,
function_name: str | None,
tool_call_id: str | None,
) -> dict[str, Any]:
return {
"type": RealtimeFeedbackType.FUNCTION_CALL_START.value,
"payload": {
"function_name": function_name,
"tool_call_id": tool_call_id,
},
}
def serialize_realtime_feedback_tool_result(result: Any) -> str | None:
"""Normalize function-call results to the string shape stored in logs."""
if result is None:
return None
return str(result)
def build_function_call_end_event(
*,
function_name: str | None,
tool_call_id: str | None,
result: Any,
) -> dict[str, Any]:
return {
"type": RealtimeFeedbackType.FUNCTION_CALL_END.value,
"payload": {
"function_name": function_name,
"tool_call_id": tool_call_id,
"result": serialize_realtime_feedback_tool_result(result),
},
}
def build_ttfb_metric_event(
*,
ttfb_seconds: float,
processor: str | None,
model: str | None,
) -> dict[str, Any]:
return {
"type": RealtimeFeedbackType.TTFB_METRIC.value,
"payload": {
"ttfb_seconds": ttfb_seconds,
"processor": processor,
"model": model,
},
}
def build_pipeline_error_event(
*,
error: str,
fatal: bool,
processor: str | None = None,
extra_payload: dict[str, Any] | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"error": error,
"fatal": fatal,
}
if processor is not None:
payload["processor"] = processor
if extra_payload:
payload.update(extra_payload)
return {
"type": RealtimeFeedbackType.PIPELINE_ERROR.value,
"payload": payload,
}
def stamp_realtime_feedback_event(
event: dict[str, Any],
*,
timestamp: str | None = None,
turn: int | None = None,
node_id: str | None = None,
node_name: str | None = None,
) -> dict[str, Any]:
stamped = dict(event)
if timestamp is not None:
stamped["timestamp"] = timestamp
if turn is not None:
stamped["turn"] = turn
if node_id is not None:
stamped["node_id"] = node_id
if node_name is not None:
stamped["node_name"] = node_name
return stamped
def realtime_feedback_event_sort_key(event: dict[str, Any]) -> str:
payload_timestamp = (event.get("payload") or {}).get("timestamp")
return payload_timestamp or event.get("timestamp") or ""

140
api/services/pipecat/realtime_feedback_observer.py
View file

@ -27,6 +27,15 @@ from typing import TYPE_CHECKING, Awaitable, Callable, Optional, Set
from loguru import logger
from api.services.pipecat.realtime_feedback_events import (
build_bot_text_event,
build_function_call_end_event,
build_function_call_start_event,
build_pipeline_error_event,
build_ttfb_metric_event,
build_user_transcription_event,
)
if TYPE_CHECKING:
from api.services.pipecat.in_memory_buffers import InMemoryLogsBuffer
@ -211,29 +220,23 @@ class RealtimeFeedbackObserver(BaseObserver):
# Handle user transcriptions (interim) - WebSocket only
elif isinstance(frame, InterimTranscriptionFrame):
await self._send_ws(
{
"type": RealtimeFeedbackType.USER_TRANSCRIPTION.value,
"payload": {
"text": frame.text,
"final": False,
"user_id": frame.user_id,
"timestamp": frame.timestamp,
},
}
build_user_transcription_event(
text=frame.text,
final=False,
user_id=frame.user_id,
timestamp=frame.timestamp,
)
)
# Handle user transcriptions (final) - WebSocket only
# Complete turn text is persisted via register_turn_handlers
elif isinstance(frame, TranscriptionFrame):
await self._send_ws(
{
"type": RealtimeFeedbackType.USER_TRANSCRIPTION.value,
"payload": {
"text": frame.text,
"final": True,
"user_id": frame.user_id,
"timestamp": frame.timestamp,
},
}
build_user_transcription_event(
text=frame.text,
final=True,
user_id=frame.user_id,
timestamp=frame.timestamp,
)
)
# Handle engine-queued speech (transition/tool messages) marked for
# log persistence. The downstream TTSTextFrame(s) from the TTS service
@ -241,23 +244,13 @@ class RealtimeFeedbackObserver(BaseObserver):
# to avoid word-level log entries from word-timestamp providers.
elif isinstance(frame, TTSSpeakFrame):
if getattr(frame, "persist_to_logs", False):
await self._append_to_buffer(
{
"type": RealtimeFeedbackType.BOT_TEXT.value,
"payload": {"text": frame.text},
}
)
await self._append_to_buffer(build_bot_text_event(text=frame.text))
# Handle bot TTS text - respect pts timing, WebSocket only
# Complete turn text is persisted via register_turn_handlers,
# except for frames explicitly flagged persist_to_logs (e.g. recording
# transcripts from play_audio) which bypass the aggregator path.
elif isinstance(frame, TTSTextFrame):
message = {
"type": RealtimeFeedbackType.BOT_TEXT.value,
"payload": {
"text": frame.text,
},
}
message = build_bot_text_event(text=frame.text)
# If frame has pts, queue it for timed delivery
if frame.pts:
@ -280,13 +273,10 @@ class RealtimeFeedbackObserver(BaseObserver):
and frame_direction == FrameDirection.DOWNSTREAM
):
await self._send_message(
{
"type": RealtimeFeedbackType.FUNCTION_CALL_START.value,
"payload": {
"function_name": frame.function_name,
"tool_call_id": frame.tool_call_id,
},
}
build_function_call_start_event(
function_name=frame.function_name,
tool_call_id=frame.tool_call_id,
)
)
# Handle function call result
elif (
@ -294,14 +284,11 @@ class RealtimeFeedbackObserver(BaseObserver):
and frame_direction == FrameDirection.DOWNSTREAM
):
await self._send_message(
{
"type": RealtimeFeedbackType.FUNCTION_CALL_END.value,
"payload": {
"function_name": frame.function_name,
"tool_call_id": frame.tool_call_id,
"result": str(frame.result) if frame.result else None,
},
}
build_function_call_end_event(
function_name=frame.function_name,
tool_call_id=frame.tool_call_id,
result=frame.result,
)
)
# Handle TTFB metrics - capture LLM generation time only
elif isinstance(frame, MetricsFrame):
@ -311,47 +298,42 @@ class RealtimeFeedbackObserver(BaseObserver):
# Only send TTFB if it's from an LLM processor
if metric_data.processor and "LLM" in metric_data.processor:
await self._send_message(
{
"type": RealtimeFeedbackType.TTFB_METRIC.value,
"payload": {
"ttfb_seconds": metric_data.value,
"processor": metric_data.processor,
"model": metric_data.model,
},
}
build_ttfb_metric_event(
ttfb_seconds=metric_data.value,
processor=metric_data.processor,
model=metric_data.model,
)
)
# Handle pipeline errors
elif isinstance(frame, ErrorFrame):
processor_name = str(frame.processor) if frame.processor else None
payload = {
"error": frame.error,
"fatal": frame.fatal,
"processor": processor_name,
}
extra_payload: dict[str, object] = {}
# Surface structured fields when the underlying exception carries
# them (e.g. google.genai APIError: code=1008, status=None,
# message="Your project has been denied access...").
exc = frame.exception
if exc is not None:
exc_type = type(exc).__name__
payload["exception_type"] = exc_type
payload["exception_message"] = str(exc)
extra_payload["exception_type"] = exc_type
extra_payload["exception_message"] = str(exc)
for attr in ("code", "status", "message", "details"):
value = getattr(exc, attr, None)
if value is None or attr in payload:
if value is None or attr in extra_payload:
continue
try:
# Ensure the value is JSON-serializable; fall back
# to str() for opaque objects (e.g. raw response).
json.dumps(value)
payload[attr] = value
extra_payload[attr] = value
except (TypeError, ValueError):
payload[attr] = str(value)
extra_payload[attr] = str(value)
await self._send_message(
{
"type": RealtimeFeedbackType.PIPELINE_ERROR.value,
"payload": payload,
}
build_pipeline_error_event(
error=frame.error,
fatal=frame.fatal,
processor=processor_name,
extra_payload=extra_payload or None,
)
)
async def _send_ws(self, message: dict):
@ -401,14 +383,11 @@ def register_turn_log_handlers(
logs_buffer.increment_turn()
try:
await logs_buffer.append(
{
"type": RealtimeFeedbackType.USER_TRANSCRIPTION.value,
"payload": {
"text": message.content,
"final": True,
"timestamp": message.timestamp,
},
}
build_user_transcription_event(
text=message.content,
final=True,
timestamp=message.timestamp,
)
)
except Exception as e:
logger.error(f"Failed to append user turn to logs buffer: {e}")
@ -418,13 +397,10 @@ def register_turn_log_handlers(
if message.content:
try:
await logs_buffer.append(
{
"type": RealtimeFeedbackType.BOT_TEXT.value,
"payload": {
"text": message.content,
"timestamp": message.timestamp,
},
}
build_bot_text_event(
text=message.content,
timestamp=message.timestamp,
)
)
except Exception as e:
logger.error(f"Failed to append assistant turn to logs buffer: {e}")

20
api/services/pipecat/run_pipeline.py
View file

@ -28,6 +28,9 @@ from api.services.pipecat.pipeline_engine_callbacks_processor import (
)
from api.services.pipecat.pipeline_metrics_aggregator import PipelineMetricsAggregator
from api.services.pipecat.pre_call_fetch import execute_pre_call_fetch
from api.services.pipecat.realtime_feedback_events import (
build_node_transition_event,
)
from api.services.pipecat.realtime_feedback_observer import (
RealtimeFeedbackObserver,
register_turn_log_handlers,
@ -465,16 +468,13 @@ async def _run_pipeline(
# Update current node on the buffer so subsequent events are tagged
in_memory_logs_buffer.set_current_node(node_id, node_name)
message = {
"type": RealtimeFeedbackType.NODE_TRANSITION.value,
"payload": {
"node_id": node_id,
"node_name": node_name,
"previous_node_id": previous_node_id,
"previous_node_name": previous_node_name,
"allow_interrupt": allow_interrupt,
},
}
message = build_node_transition_event(
node_id=node_id,
node_name=node_name,
previous_node_id=previous_node_id,
previous_node_name=previous_node_name,
allow_interrupt=allow_interrupt,
)
# Send via WebSocket if available
if ws_sender:
try:

38
api/services/pipecat/tracing_config.py
View file

@ -254,6 +254,44 @@ async def handle_langfuse_sync(event):
unregister_org_langfuse_credentials(org_id)
def build_remote_parent_context(trace_id: str | None):
"""Build an OTEL context whose active span carries ``trace_id``.
Spans started under the returned context join the Langfuse trace identified
by ``trace_id`` (Langfuse groups observations by trace id). The parent span
id is a non-existent placeholder, so spans created under it attach at the
trace root rather than nesting under a real parent span.
This is the shared primitive behind both post-call QA tracing and text-chat
trace stitching. Returns the context, or ``None`` when tracing is
unavailable or ``trace_id`` is missing/invalid.
"""
if not trace_id:
return None
if not ensure_tracing():
return None
try:
from opentelemetry.trace import (
NonRecordingSpan,
SpanContext,
TraceFlags,
set_span_in_context,
)
parent_span_context = SpanContext(
trace_id=int(trace_id, 16),
span_id=0x1,
is_remote=True,
trace_flags=TraceFlags(0x01),
)
return set_span_in_context(NonRecordingSpan(parent_span_context))
except Exception as e:
logger.warning(
f"Failed to build remote parent context for trace {trace_id}: {e}"
)
return None
def get_trace_url(trace_id: str, org_id=None) -> str | None:
"""Build a Langfuse trace URL, using org-specific host when available."""
if org_id is None:

170
api/services/pricing/workflow_run_cost.py
View file

@ -63,6 +63,95 @@ async def _update_organization_usage(
)
async def _get_pricing_organization(workflow_run):
workflow = getattr(workflow_run, "workflow", None)
organization_id = getattr(workflow, "organization_id", None)
if organization_id is None and workflow and workflow.user:
organization_id = workflow.user.selected_organization_id
if organization_id is None:
return None
return await db_client.get_organization_by_id(organization_id)
async def build_workflow_run_cost_info(workflow_run) -> dict | None:
workflow_usage_info = workflow_run.usage_info
if not workflow_usage_info:
logger.warning("No usage info available for workflow run")
return None
# Calculate cost breakdown
cost_breakdown = cost_calculator.calculate_total_cost(workflow_usage_info)
# Fetch telephony call cost
try:
telephony_cost = await _fetch_telephony_cost(workflow_run)
if telephony_cost:
telephony_cost_usd = telephony_cost["cost_usd"]
provider_name = telephony_cost["provider_name"]
cost_breakdown["telephony_call"] = telephony_cost_usd
cost_breakdown[f"{provider_name}_call"] = telephony_cost_usd
cost_breakdown["total"] = (
float(cost_breakdown["total"]) + telephony_cost_usd
)
except Exception as e:
logger.error(f"Failed to fetch telephony call cost: {e}")
# Don't fail the whole cost calculation if telephony API fails
# Convert USD to Dograh Tokens (1 cent = 1 token)
dograh_tokens = round(float(cost_breakdown["total"]) * 100, 2)
# Get organization to check if it has USD pricing
org = await _get_pricing_organization(workflow_run)
charge_usd = None
# Calculate USD cost if organization has pricing configured
if org and org.price_per_second_usd:
duration_seconds = workflow_usage_info.get("call_duration_seconds", 0)
charge_usd = duration_seconds * org.price_per_second_usd
cost_info = {
**(workflow_run.cost_info or {}),
"cost_breakdown": cost_breakdown,
"total_cost_usd": float(cost_breakdown["total"]),
"dograh_token_usage": dograh_tokens,
"calculated_at": workflow_run.created_at.isoformat(),
"call_duration_seconds": workflow_usage_info.get("call_duration_seconds", 0),
}
# Add USD cost if available
if charge_usd is not None:
cost_info["charge_usd"] = charge_usd
cost_info["price_per_second_usd"] = org.price_per_second_usd
return cost_info
async def save_workflow_run_cost_info(
workflow_run_id: int, cost_info: dict | None
) -> None:
if cost_info is None:
return
await db_client.update_workflow_run(run_id=workflow_run_id, cost_info=cost_info)
async def apply_workflow_run_usage_to_organization(
workflow_run, cost_info: dict | None
) -> None:
if cost_info is None:
return
org = await _get_pricing_organization(workflow_run)
if not org:
return
await _update_organization_usage(
org,
float(cost_info.get("dograh_token_usage") or 0),
float(cost_info.get("call_duration_seconds") or 0),
cost_info.get("charge_usd"),
)
async def calculate_workflow_run_cost(workflow_run_id: int):
logger.debug("Calculating cost for workflow run")
@ -71,85 +160,28 @@ async def calculate_workflow_run_cost(workflow_run_id: int):
logger.warning("Workflow run not found")
return
workflow_usage_info = workflow_run.usage_info
if not workflow_usage_info:
logger.warning("No usage info available for workflow run")
return
try:
# Calculate cost breakdown
cost_breakdown = cost_calculator.calculate_total_cost(workflow_usage_info)
cost_info = await build_workflow_run_cost_info(workflow_run)
if cost_info is None:
return
await save_workflow_run_cost_info(workflow_run_id, cost_info)
# Fetch telephony call cost
try:
telephony_cost = await _fetch_telephony_cost(workflow_run)
if telephony_cost:
telephony_cost_usd = telephony_cost["cost_usd"]
provider_name = telephony_cost["provider_name"]
cost_breakdown["telephony_call"] = telephony_cost_usd
cost_breakdown[f"{provider_name}_call"] = telephony_cost_usd
cost_breakdown["total"] = (
float(cost_breakdown["total"]) + telephony_cost_usd
)
await apply_workflow_run_usage_to_organization(workflow_run, cost_info)
except Exception as e:
logger.error(f"Failed to fetch telephony call cost: {e}")
# Don't fail the whole cost calculation if telephony API fails
# Store cost information back to the workflow run
# Convert USD to Dograh Tokens (1 cent = 1 token)
dograh_tokens = round(float(cost_breakdown["total"]) * 100, 2)
# Get organization to check if it has USD pricing
org = None
charge_usd = None
if (
workflow_run.workflow
and workflow_run.workflow.user
and workflow_run.workflow.user.selected_organization_id
):
org = await db_client.get_organization_by_id(
workflow_run.workflow.user.selected_organization_id
)
# Calculate USD cost if organization has pricing configured
if org and org.price_per_second_usd:
duration_seconds = workflow_usage_info.get("call_duration_seconds", 0)
charge_usd = duration_seconds * org.price_per_second_usd
cost_info = {
**workflow_run.cost_info,
"cost_breakdown": cost_breakdown,
"total_cost_usd": float(cost_breakdown["total"]),
"dograh_token_usage": dograh_tokens,
"calculated_at": workflow_run.created_at.isoformat(),
"call_duration_seconds": workflow_usage_info["call_duration_seconds"],
}
# Add USD cost if available
if charge_usd is not None:
cost_info["charge_usd"] = charge_usd
cost_info["price_per_second_usd"] = org.price_per_second_usd
# Update workflow run with cost information
await db_client.update_workflow_run(run_id=workflow_run_id, cost_info=cost_info)
# Update organization usage if applicable
if org:
try:
duration_seconds = workflow_usage_info.get("call_duration_seconds", 0)
await _update_organization_usage(
org, dograh_tokens, duration_seconds, charge_usd
)
except Exception as e:
org = await _get_pricing_organization(workflow_run)
if org:
logger.error(
f"Failed to update organization usage for org {org.id}: {e}"
)
# Don't fail the whole task if usage update fails
else:
logger.error(f"Failed to update organization usage: {e}")
# Don't fail the whole cost calculation if usage update fails
logger.info(
f"Calculated cost for workflow run: ${cost_breakdown['total']:.6f} USD ({dograh_tokens} Dograh Tokens)"
f"Calculated cost for workflow run: ${cost_info['total_cost_usd']:.6f} USD ({cost_info['dograh_token_usage']} Dograh Tokens)"
)
except Exception as e:
logger.error(f"Error calculating cost for workflow run: {e}")
raise

4
api/services/workflow/pipecat_engine.py
View file

@ -534,7 +534,7 @@ class PipecatEngine:
)
await self._update_llm_context(system_prompt, functions)
async def set_node(self, node_id: str):
async def set_node(self, node_id: str, emit_transition_event: bool = True):
"""
Simplified set_node implementation according to v2 PRD.
"""
@ -557,7 +557,7 @@ class PipecatEngine:
nodes_visited.append(node.name)
# Send node transition event if callback is provided
if self._node_transition_callback:
if emit_transition_event and self._node_transition_callback:
try:
await self._node_transition_callback(
node_id,

39
api/services/workflow/qa/tracing.py
View file

@ -6,7 +6,10 @@ import re
from loguru import logger
from api.db.models import WorkflowRunModel
from api.services.pipecat.tracing_config import get_trace_url
from api.services.pipecat.tracing_config import (
build_remote_parent_context,
get_trace_url,
)
def extract_trace_id(gathered_context: dict) -> str | None:
@ -33,36 +36,12 @@ def setup_langfuse_parent_context(workflow_run: WorkflowRunModel):
Returns the parent context object, or None if tracing is unavailable.
"""
try:
from opentelemetry.trace import (
NonRecordingSpan,
SpanContext,
TraceFlags,
set_span_in_context,
)
from api.services.pipecat.tracing_config import ensure_tracing
if not ensure_tracing():
return None
gathered_context = workflow_run.gathered_context or {}
trace_id = extract_trace_id(gathered_context)
if not trace_id:
logger.debug("No trace_id found, skipping Langfuse tracing")
return None
parent_span_ctx = SpanContext(
trace_id=int(trace_id, 16),
span_id=0x1,
is_remote=True,
trace_flags=TraceFlags(0x01),
)
return set_span_in_context(NonRecordingSpan(parent_span_ctx))
except Exception as e:
logger.warning(f"Failed to set up Langfuse parent context: {e}")
gathered_context = workflow_run.gathered_context or {}
trace_id = extract_trace_id(gathered_context)
if not trace_id:
logger.debug("No trace_id found, skipping Langfuse tracing")
return None
return build_remote_parent_context(trace_id)
def add_qa_span_to_trace(

143
api/services/workflow/text_chat_logs.py Normal file
View file

@ -0,0 +1,143 @@
"""Helpers for projecting text-chat session state into run-log snapshots."""
from typing import Any
from api.services.pipecat.realtime_feedback_events import (
build_bot_text_event,
build_function_call_end_event,
build_function_call_start_event,
build_node_transition_event,
build_pipeline_error_event,
build_user_transcription_event,
realtime_feedback_event_sort_key,
stamp_realtime_feedback_event,
)
def visible_text_chat_turns(session_data: dict[str, Any]) -> list[dict[str, Any]]:
"""Return the active branch of turns for the current text-chat session.
After a rewind, `session_data["turns"]` may still contain future turns until
the next message is sent. Those turns are no longer part of the visible
branch, so callers that synthesize transcript/log views should trim at
`cursor_turn_id`.
"""
turns = list(session_data.get("turns") or [])
cursor_turn_id = session_data.get("cursor_turn_id")
if cursor_turn_id is None:
return turns
for index, turn in enumerate(turns):
if turn.get("id") == cursor_turn_id:
return turns[: index + 1]
return turns
def build_text_chat_realtime_feedback_events(
session_data: dict[str, Any],
) -> list[dict[str, Any]]:
"""Project text-chat session state into `workflow_runs.logs` event format.
`workflow_run_text_sessions` holds the authoritative rewindable conversation
state. Historical run pages and QA helpers read the normalized
`workflow_runs.logs.realtime_feedback_events` schema instead, so this helper
rebuilds that snapshot from the currently visible branch.
"""
events: list[dict[str, Any]] = []
last_emitted_node_id: str | None = None
for turn_index, turn in enumerate(visible_text_chat_turns(session_data)):
turn_events = list(turn.get("events") or [])
for event in turn_events:
payload = dict(event.get("payload") or {})
event_type = event.get("type")
timestamp = event.get("created_at") or turn.get("created_at")
if event_type == "node_transition":
node_id = payload.get("node_id")
if node_id is not None and node_id == last_emitted_node_id:
continue
snapshot_event = stamp_realtime_feedback_event(
build_node_transition_event(
node_id=node_id,
node_name=payload.get("node_name"),
previous_node_id=payload.get("previous_node_id"),
previous_node_name=payload.get("previous_node_name"),
allow_interrupt=bool(payload.get("allow_interrupt", False)),
),
timestamp=timestamp,
turn=turn_index,
node_id=node_id,
node_name=payload.get("node_name"),
)
if node_id is not None:
last_emitted_node_id = node_id
events.append(snapshot_event)
elif event_type == "tool_call_started":
events.append(
stamp_realtime_feedback_event(
build_function_call_start_event(
function_name=payload.get("function_name"),
tool_call_id=payload.get("tool_call_id"),
),
timestamp=timestamp,
turn=turn_index,
)
)
elif event_type == "tool_call_result":
events.append(
stamp_realtime_feedback_event(
build_function_call_end_event(
function_name=payload.get("function_name"),
tool_call_id=payload.get("tool_call_id"),
result=payload.get("result"),
),
timestamp=timestamp,
turn=turn_index,
)
)
elif event_type == "execution_error":
events.append(
stamp_realtime_feedback_event(
build_pipeline_error_event(
error=payload.get("message", "Execution error"),
fatal=True,
),
timestamp=timestamp,
turn=turn_index,
)
)
user_message = turn.get("user_message") or {}
if user_message.get("text"):
message_timestamp = user_message.get("created_at") or turn.get("created_at")
events.append(
stamp_realtime_feedback_event(
build_user_transcription_event(
text=user_message["text"],
final=True,
timestamp=message_timestamp,
),
timestamp=message_timestamp,
turn=turn_index,
)
)
assistant_message = turn.get("assistant_message") or {}
if assistant_message.get("text"):
message_timestamp = assistant_message.get("created_at") or turn.get(
"created_at"
)
events.append(
stamp_realtime_feedback_event(
build_bot_text_event(
text=assistant_message["text"],
timestamp=message_timestamp,
),
timestamp=message_timestamp,
turn=turn_index,
)
)
return sorted(events, key=realtime_feedback_event_sort_key)

57
api/services/workflow/text_chat_runner.py
View file

@ -1,4 +1,5 @@
import asyncio
import hashlib
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime
@ -28,6 +29,7 @@ from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.run_context import set_current_org_id
from api.db import db_client
from api.enums import WorkflowRunMode, WorkflowRunState
@ -39,6 +41,10 @@ from api.services.pipecat.pipeline_metrics_aggregator import (
)
from api.services.pipecat.recording_audio_cache import create_recording_audio_fetcher
from api.services.pipecat.service_factory import create_llm_service
from api.services.pipecat.tracing_config import (
build_remote_parent_context,
get_trace_url,
)
from api.services.workflow.dto import ReactFlowDTO
from api.services.workflow.pipecat_engine import PipecatEngine
from api.services.workflow.workflow_graph import WorkflowGraph
@ -49,6 +55,19 @@ TEXT_CHAT_IDLE_SETTLE_SECONDS = 0.2
TEXT_CHAT_INTERNAL_CANCEL_REASON = "text_chat_turn_complete"
def text_chat_trace_id(workflow_run_id: int) -> str:
"""Deterministic Langfuse trace id for a text-chat session.
Each turn runs in its own short-lived pipeline, so there is no single
long-running task to own the trace the way a voice call does. Deriving the
id from the run id means every turn re-creates the *same* trace id and all
per-turn spans land in one shared trace — without persisting extra state
across the otherwise stateless turn requests.
"""
digest = hashlib.sha256(f"dograh-text-chat:{workflow_run_id}".encode()).hexdigest()
return digest[:32]
def default_text_chat_checkpoint() -> dict[str, Any]:
return {
"version": TEXT_CHAT_CHECKPOINT_VERSION,
@ -379,6 +398,12 @@ async def execute_text_chat_pending_turn(
if workflow is None:
raise ValueError("Workflow not found for text chat execution")
# Stamp the async context so OTEL spans are tagged with this org and routed
# to its Langfuse project (the voice paths do this in run_pipeline /
# webrtc_signaling; the text path previously skipped it, so its spans never
# reached org-specific exporters).
set_current_org_id(workflow.organization_id)
run_definition = workflow_run.definition
run_configs = run_definition.workflow_configurations or {}
@ -482,6 +507,17 @@ async def execute_text_chat_pending_turn(
audio_config = create_audio_config(WorkflowRunMode.SMALLWEBRTC.value)
pipeline_metrics_aggregator = PipelineMetricsAggregator()
# Stitch every per-turn pipeline of this session into one Langfuse trace by
# handing each task the same remote parent context (derived from the run id).
trace_id = text_chat_trace_id(workflow_run_id)
conversation_parent_context = build_remote_parent_context(trace_id)
# The stitched trace has no real root span (each per-turn conversation span
# hangs off a synthetic remote parent), so Langfuse can't infer a name and
# shows "Unnamed trace". Name it explicitly via the conversation span.
trace_span_attributes = {
"langfuse.trace.name": workflow_run.name or f"text-chat-{workflow_run_id}"
}
pipeline = Pipeline(
[
llm,
@ -490,7 +526,14 @@ async def execute_text_chat_pending_turn(
pipeline_metrics_aggregator,
]
)
task = create_pipeline_task(pipeline, workflow_run_id, audio_config)
task = create_pipeline_task(
pipeline,
workflow_run_id,
audio_config,
conversation_parent_context=conversation_parent_context,
conversation_type="text",
additional_span_attributes=trace_span_attributes,
)
runner = PipelineRunner(handle_sigint=False, handle_sigterm=False)
runner_task = asyncio.create_task(runner.run(task))
@ -511,7 +554,10 @@ async def execute_text_chat_pending_turn(
current_node_id = base_checkpoint.get("current_node_id")
target_node_id = current_node_id or workflow_graph.start_node_id
await engine.set_node(target_node_id)
await engine.set_node(
target_node_id,
emit_transition_event=current_node_id is None,
)
opening_marker = capture_processor.activity_count
opening_expects_llm = pending_user_message is None and (
@ -581,13 +627,18 @@ async def execute_text_chat_pending_turn(
"tool_state": jsonable_encoder(base_checkpoint.get("tool_state") or {}),
}
encoded_gathered_context = jsonable_encoder(gathered_context)
trace_url = get_trace_url(trace_id, org_id=workflow.organization_id)
if trace_url:
encoded_gathered_context = {**encoded_gathered_context, "trace_url": trace_url}
return TextChatTurnExecutionResult(
assistant_text=assistant_text,
assistant_created_at=assistant_created_at,
events=jsonable_encoder(capture_processor.events),
usage=jsonable_encoder(usage),
checkpoint=updated_checkpoint,
gathered_context=jsonable_encoder(gathered_context),
gathered_context=encoded_gathered_context,
initial_context=jsonable_encoder(initial_context),
state=(
WorkflowRunState.COMPLETED.value

396
api/services/workflow/text_chat_session_service.py Normal file
View file

@ -0,0 +1,396 @@
"""Service helpers for text-chat session lifecycle orchestration."""
from datetime import UTC, datetime
from typing import Any
from uuid import uuid4
from api.db import db_client
from api.db.models import WorkflowRunTextSessionModel
from api.db.workflow_run_text_session_client import (
WorkflowRunTextSessionRevisionConflictError,
)
from api.services.pricing.workflow_run_cost import build_workflow_run_cost_info
from api.services.workflow.text_chat_logs import (
build_text_chat_realtime_feedback_events,
)
from api.services.workflow.text_chat_runner import (
default_text_chat_checkpoint,
execute_text_chat_pending_turn,
merge_text_chat_usage_info,
normalize_text_chat_checkpoint,
)
TEXT_CHAT_SESSION_VERSION = 1
class TextChatSessionRevisionConflictError(Exception):
def __init__(self, expected_revision: int, actual_revision: int):
self.expected_revision = expected_revision
self.actual_revision = actual_revision
super().__init__(
"Text chat session revision conflict: "
f"expected {expected_revision}, found {actual_revision}"
)
class TextChatSessionExecutionError(Exception):
"""Raised when the assistant turn fails to execute."""
class TextChatPendingTurnLostError(Exception):
"""Raised when the pending turn disappears before persistence completes."""
class TextChatTurnNotFoundError(Exception):
"""Raised when a requested rewind cursor does not exist in the session."""
def default_text_chat_session_data() -> dict[str, Any]:
return {
"version": TEXT_CHAT_SESSION_VERSION,
"status": "idle",
"cursor_turn_id": None,
"turns": [],
"discarded_future": [],
"simulator": {
"enabled": False,
"config": {},
},
}
def normalize_text_chat_session_data(
session_data: dict[str, Any] | None,
) -> dict[str, Any]:
normalized = {
**default_text_chat_session_data(),
**(session_data or {}),
}
normalized["turns"] = list(normalized.get("turns") or [])
normalized["discarded_future"] = list(normalized.get("discarded_future") or [])
simulator = normalized.get("simulator") or {}
normalized["simulator"] = {
"enabled": bool(simulator.get("enabled", False)),
"config": dict(simulator.get("config") or {}),
}
return normalized
async def initialize_text_chat_session(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
) -> WorkflowRunTextSessionModel:
session_data = normalize_text_chat_session_data(text_session.session_data)
checkpoint = normalize_text_chat_checkpoint(text_session.checkpoint)
session_data["turns"] = [build_pending_text_chat_turn(user_text=None)]
session_data["status"] = "pending_assistant_turn"
checkpoint["anchor_turn_id"] = latest_completed_text_chat_turn_id(
session_data["turns"]
)
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
checkpoint=checkpoint,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
return await _reload_text_chat_session(run_id, text_session)
async def append_text_chat_user_message(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
user_text: str,
expected_revision: int | None,
) -> WorkflowRunTextSessionModel:
session_data = normalize_text_chat_session_data(text_session.session_data)
checkpoint = normalize_text_chat_checkpoint(text_session.checkpoint)
active_turns, discarded_future = truncate_text_chat_future_turns(session_data)
active_turns.append(build_pending_text_chat_turn(user_text=user_text))
session_data["turns"] = active_turns
session_data["discarded_future"] = discarded_future
session_data["cursor_turn_id"] = None
session_data["status"] = "pending_assistant_turn"
checkpoint["anchor_turn_id"] = latest_completed_text_chat_turn_id(active_turns)
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
checkpoint=checkpoint,
expected_revision=expected_revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
return await _reload_text_chat_session(run_id, text_session)
async def rewind_text_chat_session_state(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
cursor_turn_id: str | None,
expected_revision: int | None,
) -> WorkflowRunTextSessionModel:
session_data = normalize_text_chat_session_data(text_session.session_data)
validate_text_chat_turn_cursor(session_data, cursor_turn_id)
session_data["cursor_turn_id"] = cursor_turn_id
session_data["status"] = "rewound" if cursor_turn_id else "idle"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=session_data,
expected_revision=expected_revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
await db_client.update_workflow_run(
run_id,
logs={
"realtime_feedback_events": build_text_chat_realtime_feedback_events(
session_data
)
},
)
return await _reload_text_chat_session(run_id, text_session)
async def execute_pending_text_chat_turn(
*,
workflow_id: int,
run_id: int,
text_session: WorkflowRunTextSessionModel,
) -> WorkflowRunTextSessionModel:
"""Execute the current pending assistant turn and persist its side effects."""
session_data = normalize_text_chat_session_data(text_session.session_data)
checkpoint = normalize_text_chat_checkpoint(text_session.checkpoint)
try:
execution = await execute_text_chat_pending_turn(
workflow_run_id=run_id,
workflow_id=workflow_id,
session_data=session_data,
checkpoint=checkpoint,
)
except Exception as e:
await _mark_pending_turn_failed(
run_id=run_id,
text_session=text_session,
error_message=str(e),
)
raise TextChatSessionExecutionError(
"Failed to execute text chat assistant turn"
) from e
completed_session_data = normalize_text_chat_session_data(text_session.session_data)
completed_turns = list(completed_session_data.get("turns") or [])
if not completed_turns or completed_turns[-1].get("status") != "pending":
raise TextChatPendingTurnLostError(
"Text chat session lost its pending turn before completion"
)
completed_turns[-1]["status"] = "completed"
completed_turns[-1]["assistant_message"] = (
{
"text": execution.assistant_text,
"created_at": execution.assistant_created_at,
}
if execution.assistant_text
else None
)
completed_turns[-1]["events"] = execution.events
completed_turns[-1]["usage"] = execution.usage
completed_turns[-1]["checkpoint_after_turn"] = execution.checkpoint
completed_session_data["turns"] = completed_turns
completed_session_data["status"] = "idle"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=completed_session_data,
checkpoint=execution.checkpoint,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError as e:
raise TextChatSessionRevisionConflictError(
expected_revision=e.expected_revision,
actual_revision=e.actual_revision,
) from e
existing_usage_info = text_session.workflow_run.usage_info or {}
merged_usage_info = merge_text_chat_usage_info(existing_usage_info, execution.usage)
text_chat_logs = {
"realtime_feedback_events": build_text_chat_realtime_feedback_events(
completed_session_data
)
}
await db_client.update_workflow_run(
run_id,
initial_context=execution.initial_context,
usage_info=merged_usage_info,
gathered_context=execution.gathered_context,
logs=text_chat_logs,
state=execution.state,
is_completed=execution.is_completed,
)
workflow_run = await db_client.get_workflow_run_by_id(run_id)
if workflow_run:
cost_info = await build_workflow_run_cost_info(workflow_run)
if cost_info is not None:
await db_client.update_workflow_run(run_id, cost_info=cost_info)
return await _reload_text_chat_session(run_id, text_session)
def validate_text_chat_turn_cursor(
session_data: dict[str, Any],
cursor_turn_id: str | None,
) -> None:
if cursor_turn_id is None:
return
if not any(turn.get("id") == cursor_turn_id for turn in session_data["turns"]):
raise TextChatTurnNotFoundError("Turn not found in text chat session")
def truncate_text_chat_future_turns(
session_data: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
cursor_turn_id = session_data.get("cursor_turn_id")
turns = list(session_data.get("turns") or [])
discarded_future = list(session_data.get("discarded_future") or [])
if cursor_turn_id is None:
return turns, discarded_future
for index, turn in enumerate(turns):
if turn.get("id") == cursor_turn_id:
active_turns = turns[: index + 1]
future_turns = turns[index + 1 :]
if future_turns:
discarded_future.append(
{
"rewound_from_turn_id": cursor_turn_id,
"discarded_at": datetime.now(UTC).isoformat(),
"turns": future_turns,
}
)
return active_turns, discarded_future
raise TextChatTurnNotFoundError("Turn not found in text chat session")
def latest_completed_text_chat_turn_id(turns: list[dict[str, Any]]) -> str | None:
for turn in reversed(turns):
if turn.get("status") == "completed":
return turn.get("id")
return None
def build_pending_text_chat_turn(*, user_text: str | None) -> dict[str, Any]:
now = datetime.now(UTC).isoformat()
return {
"id": f"turn_{uuid4().hex[:12]}",
"status": "pending",
"created_at": now,
"user_message": (
{
"text": user_text,
"created_at": now,
}
if user_text is not None
else None
),
"assistant_message": None,
"events": [],
"usage": {},
}
async def _mark_pending_turn_failed(
*,
run_id: int,
text_session: WorkflowRunTextSessionModel,
error_message: str,
) -> None:
failed_session_data = normalize_text_chat_session_data(text_session.session_data)
failed_turns = list(failed_session_data.get("turns") or [])
if not failed_turns or failed_turns[-1].get("status") != "pending":
return
failed_turns[-1]["status"] = "failed"
failed_turns[-1]["events"] = [
*(failed_turns[-1].get("events") or []),
{
"type": "execution_error",
"created_at": datetime.now(UTC).isoformat(),
"payload": {"message": error_message},
},
]
failed_session_data["turns"] = failed_turns
failed_session_data["status"] = "error"
try:
await db_client.update_workflow_run_text_session(
run_id,
session_data=failed_session_data,
expected_revision=text_session.revision,
)
except WorkflowRunTextSessionRevisionConflictError:
return
async def _reload_text_chat_session(
run_id: int,
text_session: WorkflowRunTextSessionModel,
) -> WorkflowRunTextSessionModel:
organization_id = text_session.workflow_run.workflow.organization_id
updated_text_session = await db_client.get_workflow_run_text_session(
run_id,
organization_id=organization_id,
)
if updated_text_session is None:
raise TextChatSessionExecutionError("Text chat session not found after update")
return updated_text_session
__all__ = [
"TEXT_CHAT_SESSION_VERSION",
"TextChatTurnNotFoundError",
"append_text_chat_user_message",
"build_pending_text_chat_turn",
"TextChatPendingTurnLostError",
"TextChatSessionExecutionError",
"TextChatSessionRevisionConflictError",
"default_text_chat_checkpoint",
"default_text_chat_session_data",
"execute_pending_text_chat_turn",
"initialize_text_chat_session",
"latest_completed_text_chat_turn_id",
"normalize_text_chat_checkpoint",
"normalize_text_chat_session_data",
"rewind_text_chat_session_state",
"truncate_text_chat_future_turns",
"validate_text_chat_turn_cursor",
]

53
api/tests/test_realtime_feedback_events.py Normal file
View file

@ -0,0 +1,53 @@
from api.services.pipecat.realtime_feedback_events import (
build_bot_text_event,
build_function_call_end_event,
build_node_transition_event,
realtime_feedback_event_sort_key,
stamp_realtime_feedback_event,
)
def test_build_function_call_end_event_serializes_results():
event = build_function_call_end_event(
function_name="lookup_contact",
tool_call_id="tool-1",
result={"contact_id": 42},
)
assert event == {
"type": "rtf-function-call-end",
"payload": {
"function_name": "lookup_contact",
"tool_call_id": "tool-1",
"result": "{'contact_id': 42}",
},
}
def test_stamp_and_sort_realtime_feedback_events():
node_transition = stamp_realtime_feedback_event(
build_node_transition_event(
node_id="node-1",
node_name="Greeting",
previous_node_id=None,
previous_node_name=None,
),
timestamp="2026-01-01T00:00:03+00:00",
turn=0,
node_id="node-1",
node_name="Greeting",
)
bot_text = stamp_realtime_feedback_event(
build_bot_text_event(
text="Hello there",
timestamp="2026-01-01T00:00:01+00:00",
),
timestamp="2026-01-01T00:00:02+00:00",
turn=0,
)
events = sorted([node_transition, bot_text], key=realtime_feedback_event_sort_key)
assert events == [bot_text, node_transition]
assert node_transition["node_id"] == "node-1"
assert node_transition["node_name"] == "Greeting"

126
api/tests/test_text_chat_logs.py Normal file
View file

@ -0,0 +1,126 @@
from api.services.workflow.text_chat_logs import (
build_text_chat_realtime_feedback_events,
visible_text_chat_turns,
)
def test_visible_text_chat_turns_trims_to_cursor_branch():
session_data = {
"cursor_turn_id": "turn-2",
"turns": [
{"id": "turn-1"},
{"id": "turn-2"},
{"id": "turn-3"},
],
}
assert visible_text_chat_turns(session_data) == [
{"id": "turn-1"},
{"id": "turn-2"},
]
def test_build_text_chat_realtime_feedback_events_uses_visible_branch_and_dedupes_node_transitions():
session_data = {
"cursor_turn_id": "turn-2",
"turns": [
{
"id": "turn-1",
"created_at": "2026-01-01T00:00:00+00:00",
"events": [
{
"type": "node_transition",
"created_at": "2026-01-01T00:00:00+00:00",
"payload": {
"node_id": "node-start",
"node_name": "Start",
"previous_node_id": None,
"previous_node_name": None,
"allow_interrupt": False,
},
}
],
"user_message": None,
"assistant_message": {
"text": "Hello",
"created_at": "2026-01-01T00:00:01+00:00",
},
},
{
"id": "turn-2",
"created_at": "2026-01-01T00:00:02+00:00",
"events": [
{
"type": "node_transition",
"created_at": "2026-01-01T00:00:02+00:00",
"payload": {
"node_id": "node-start",
"node_name": "Start",
"previous_node_id": None,
"previous_node_name": None,
"allow_interrupt": False,
},
},
{
"type": "tool_call_started",
"created_at": "2026-01-01T00:00:03+00:00",
"payload": {
"function_name": "lookup_contact",
"tool_call_id": "tool-1",
},
},
{
"type": "tool_call_result",
"created_at": "2026-01-01T00:00:04+00:00",
"payload": {
"function_name": "lookup_contact",
"tool_call_id": "tool-1",
"result": {"contact_id": 42},
},
},
],
"user_message": {
"text": "Find Abhishek",
"created_at": "2026-01-01T00:00:02+00:00",
},
"assistant_message": {
"text": "I found one match.",
"created_at": "2026-01-01T00:00:05+00:00",
},
},
{
"id": "turn-3",
"created_at": "2026-01-01T00:00:06+00:00",
"events": [
{
"type": "execution_error",
"created_at": "2026-01-01T00:00:06+00:00",
"payload": {"message": "Should be hidden after rewind"},
}
],
"user_message": {
"text": "This turn is rewound away",
"created_at": "2026-01-01T00:00:06+00:00",
},
"assistant_message": None,
},
],
}
events = build_text_chat_realtime_feedback_events(session_data)
assert [event["type"] for event in events] == [
"rtf-node-transition",
"rtf-bot-text",
"rtf-user-transcription",
"rtf-function-call-start",
"rtf-function-call-end",
"rtf-bot-text",
]
assert events[0]["payload"]["node_name"] == "Start"
assert events[2]["payload"]["text"] == "Find Abhishek"
assert events[4]["payload"]["result"] == "{'contact_id': 42}"
assert all(
event.get("payload", {}).get("error") != "Should be hidden after rewind"
for event in events
)

45
api/tests/test_text_chat_session_service.py Normal file
View file

@ -0,0 +1,45 @@
import pytest
from api.services.workflow.text_chat_session_service import (
TextChatTurnNotFoundError,
build_pending_text_chat_turn,
truncate_text_chat_future_turns,
validate_text_chat_turn_cursor,
)
def test_build_pending_text_chat_turn_sets_pending_shape():
turn = build_pending_text_chat_turn(user_text="Hello")
assert turn["id"].startswith("turn_")
assert turn["status"] == "pending"
assert turn["user_message"]["text"] == "Hello"
assert turn["assistant_message"] is None
assert turn["events"] == []
assert turn["usage"] == {}
def test_truncate_text_chat_future_turns_moves_rewound_branch_to_discarded_future():
session_data = {
"cursor_turn_id": "turn-2",
"turns": [
{"id": "turn-1"},
{"id": "turn-2"},
{"id": "turn-3"},
],
"discarded_future": [],
}
active_turns, discarded_future = truncate_text_chat_future_turns(session_data)
assert active_turns == [{"id": "turn-1"}, {"id": "turn-2"}]
assert discarded_future[0]["rewound_from_turn_id"] == "turn-2"
assert discarded_future[0]["turns"] == [{"id": "turn-3"}]
def test_validate_text_chat_turn_cursor_raises_for_missing_turn():
with pytest.raises(TextChatTurnNotFoundError):
validate_text_chat_turn_cursor(
{"turns": [{"id": "turn-1"}]},
"turn-404",
)

87
api/tests/test_workflow_run_cost.py Normal file
View file

@ -0,0 +1,87 @@
from datetime import UTC, datetime
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from api.services.pricing import workflow_run_cost as workflow_run_cost_mod
from api.services.pricing.workflow_run_cost import (
build_workflow_run_cost_info,
calculate_workflow_run_cost,
)
def _make_workflow_run():
return SimpleNamespace(
id=123,
workflow_id=456,
mode="textchat",
created_at=datetime.now(UTC),
usage_info={
"llm": {},
"tts": {},
"stt": {},
"call_duration_seconds": 7,
},
cost_info={},
workflow=SimpleNamespace(
organization_id=42,
user=SimpleNamespace(selected_organization_id=42),
),
)
@pytest.mark.asyncio
async def test_build_workflow_run_cost_info_does_not_update_org_usage(monkeypatch):
workflow_run = _make_workflow_run()
get_org = AsyncMock(return_value=SimpleNamespace(id=42, price_per_second_usd=1.5))
update_usage = AsyncMock()
monkeypatch.setattr(
workflow_run_cost_mod.db_client, "get_organization_by_id", get_org
)
monkeypatch.setattr(
workflow_run_cost_mod.db_client, "update_usage_after_run", update_usage
)
cost_info = await build_workflow_run_cost_info(workflow_run)
assert cost_info is not None
assert cost_info["call_duration_seconds"] == 7
assert "cost_breakdown" in cost_info
assert "dograh_token_usage" in cost_info
assert cost_info["charge_usd"] == 10.5
update_usage.assert_not_called()
@pytest.mark.asyncio
async def test_calculate_workflow_run_cost_keeps_org_usage_side_effect_in_wrapper(
monkeypatch,
):
workflow_run = _make_workflow_run()
get_org = AsyncMock(return_value=SimpleNamespace(id=42, price_per_second_usd=None))
update_run = AsyncMock()
update_usage = AsyncMock()
monkeypatch.setattr(
workflow_run_cost_mod.db_client,
"get_workflow_run_by_id",
AsyncMock(return_value=workflow_run),
)
monkeypatch.setattr(
workflow_run_cost_mod.db_client, "get_organization_by_id", get_org
)
monkeypatch.setattr(
workflow_run_cost_mod.db_client, "update_workflow_run", update_run
)
monkeypatch.setattr(
workflow_run_cost_mod.db_client, "update_usage_after_run", update_usage
)
await calculate_workflow_run_cost(workflow_run.id)
update_run.assert_awaited_once()
saved_kwargs = update_run.await_args.kwargs
assert saved_kwargs["run_id"] == workflow_run.id
assert "cost_breakdown" in saved_kwargs["cost_info"]
update_usage.assert_awaited_once()

89
api/tests/test_workflow_text_chat.py
View file

@ -8,6 +8,15 @@ from api.tests.integrations._run_pipeline_helpers import USER_CONFIGURATION
from pipecat.tests import MockLLMService
def _log_texts(logs: dict | None, event_type: str) -> list[str]:
events = (logs or {}).get("realtime_feedback_events") or []
return [
event.get("payload", {}).get("text", "")
for event in events
if event.get("type") == event_type
]
async def _create_user_and_workflow(
db_session,
async_session,
@ -115,6 +124,11 @@ async def test_text_chat_session_creation_executes_initial_assistant_turn(
)
assert create_response.status_code == 200
created = create_response.json()
run_response = await client.get(
f"/api/v1/workflow/{workflow.id}/runs/{created['workflow_run_id']}"
)
assert run_response.status_code == 200
run_payload = run_response.json()
turns = created["session_data"]["turns"]
assert created["revision"] == 2
@ -127,6 +141,16 @@ async def test_text_chat_session_creation_executes_initial_assistant_turn(
assert created["checkpoint"]["current_node_id"] == "start"
assert created["state"] == "running"
assert "Start" in (created["gathered_context"] or {}).get("nodes_visited", [])
workflow_run = await db_session.get_workflow_run_by_id(created["workflow_run_id"])
assert workflow_run is not None
assert workflow_run.cost_info[
"call_duration_seconds"
] == workflow_run.usage_info.get("call_duration_seconds", 0)
assert "cost_breakdown" in workflow_run.cost_info
assert "dograh_token_usage" in workflow_run.cost_info
assert _log_texts(run_payload["logs"], "rtf-bot-text") == [
"Hello from the workflow tester."
]
@pytest.mark.asyncio
@ -217,6 +241,11 @@ async def test_text_chat_message_executes_assistant_turn(
},
)
assert message_response.status_code == 200
run_response = await client.get(
f"/api/v1/workflow/{workflow.id}/runs/{created['workflow_run_id']}"
)
assert run_response.status_code == 200
run_payload = run_response.json()
payload = message_response.json()
turns = payload["session_data"]["turns"]
@ -232,6 +261,18 @@ async def test_text_chat_message_executes_assistant_turn(
assert payload["checkpoint"]["current_node_id"] == "start"
assert payload["state"] == "running"
assert "Start" in (payload["gathered_context"] or {}).get("nodes_visited", [])
workflow_run = await db_session.get_workflow_run_by_id(created["workflow_run_id"])
assert workflow_run is not None
assert workflow_run.cost_info[
"call_duration_seconds"
] == workflow_run.usage_info.get("call_duration_seconds", 0)
assert "cost_breakdown" in workflow_run.cost_info
assert "dograh_token_usage" in workflow_run.cost_info
assert _log_texts(run_payload["logs"], "rtf-user-transcription") == ["Hi there"]
assert _log_texts(run_payload["logs"], "rtf-bot-text") == [
"Welcome to the workflow tester.",
"Hello from the workflow tester.",
]
@pytest.mark.asyncio
@ -330,8 +371,13 @@ async def test_text_chat_executes_deferred_tool_calls_after_text_response(
},
)
assert message_response.status_code == 200
run_response = await client.get(
f"/api/v1/workflow/{workflow.id}/runs/{session['workflow_run_id']}"
)
assert run_response.status_code == 200
payload = message_response.json()
run_payload = run_response.json()
assistant_text = payload["session_data"]["turns"][1]["assistant_message"]["text"]
assert "Let me transfer you." in assistant_text
@ -342,6 +388,21 @@ async def test_text_chat_executes_deferred_tool_calls_after_text_response(
and event["payload"]["function_name"] == "go_to_agent_one"
for event in payload["session_data"]["turns"][1]["events"]
)
node_transition_names = [
event["payload"]["node_name"]
for event in run_payload["logs"]["realtime_feedback_events"]
if event["type"] == "rtf-node-transition"
]
assert node_transition_names == ["Start", "Agent One"]
function_call_event_names = [
event["type"]
for event in run_payload["logs"]["realtime_feedback_events"]
if event["type"] in {"rtf-function-call-start", "rtf-function-call-end"}
]
assert function_call_event_names == [
"rtf-function-call-start",
"rtf-function-call-end",
]
@pytest.mark.asyncio
@ -773,6 +834,11 @@ async def test_text_chat_rewind_reuses_checkpoint_snapshot(
assert rewind_response.status_code == 200
rewound = rewind_response.json()
assert rewound["session_data"]["cursor_turn_id"] == first_turn_id
rewound_run_response = await client.get(
f"/api/v1/workflow/{workflow.id}/runs/{session['workflow_run_id']}"
)
assert rewound_run_response.status_code == 200
rewound_run_payload = rewound_run_response.json()
third_message = await client.post(
f"/api/v1/workflow/{workflow.id}/text-chat/sessions/{session['workflow_run_id']}/messages",
@ -782,6 +848,11 @@ async def test_text_chat_rewind_reuses_checkpoint_snapshot(
},
)
assert third_message.status_code == 200
final_run_response = await client.get(
f"/api/v1/workflow/{workflow.id}/runs/{session['workflow_run_id']}"
)
assert final_run_response.status_code == 200
final_run_payload = final_run_response.json()
payload = third_message.json()
assert payload["checkpoint"]["current_node_id"] == "agent1"
@ -792,6 +863,24 @@ async def test_text_chat_rewind_reuses_checkpoint_snapshot(
payload["session_data"]["turns"][2]["assistant_message"]["text"]
== "Back in agent one."
)
assert _log_texts(rewound_run_payload["logs"], "rtf-user-transcription") == [
"First turn"
]
assert "Second turn" not in _log_texts(
rewound_run_payload["logs"], "rtf-user-transcription"
)
assert "Agent two here." not in _log_texts(
rewound_run_payload["logs"], "rtf-bot-text"
)
assert _log_texts(final_run_payload["logs"], "rtf-user-transcription") == [
"First turn",
"Third turn after rewind",
]
assert _log_texts(final_run_payload["logs"], "rtf-bot-text") == [
"Welcome to the rewind test.",
"Agent one here.",
"Back in agent one.",
]
@pytest.mark.asyncio

2
docs/api-reference/openapi.json
View file

File diff suppressed because one or more lines are too long

2
pipecat

@ -1 +1 @@
Subproject commit 6b4474c1b870eae5e42ef7c5eec1b7f37fcecc61
Subproject commit d1e23ca521f5412a9dc09430ada730500e15a7ab

4
sdk/python/src/dograh_sdk/_generated_models.py
View file

@ -1,6 +1,6 @@
# generated by datamodel-codegen:
# filename: dograh-openapi-XXXXXX.json.bGP2QR1Vrx
# timestamp: 2026-05-20T08:41:57+00:00
# filename: dograh-openapi-XXXXXX.json.4p15PiTCyh
# timestamp: 2026-05-21T07:00:16+00:00
from __future__ import annotations

58
ui/src/app/usage/page.tsx
View file

@ -1,6 +1,6 @@
"use client";
import { ChevronLeft, ChevronRight, Download, Globe } from 'lucide-react';
import { ArrowDownLeft, ArrowUpRight, ChevronLeft, ChevronRight, Download, Globe, MessageSquare, Phone } from 'lucide-react';
import { useRouter, useSearchParams } from 'next/navigation';
import { useCallback, useEffect, useId, useState } from 'react';
import TimezoneSelect, { type ITimezoneOption } from 'react-timezone-select';
@ -23,6 +23,7 @@ import {
TableHeader,
TableRow,
} from '@/components/ui/table';
import { Tooltip, TooltipContent, TooltipTrigger } from '@/components/ui/tooltip';
import { useUserConfig } from '@/context/UserConfigContext';
import { useAuth } from '@/lib/auth';
import { usageFilterAttributes } from '@/lib/filterAttributes';
@ -32,6 +33,53 @@ import { ActiveFilter, DateRangeValue } from '@/types/filters';
// Get local timezone
const getLocalTimezone = () => Intl.DateTimeFormat().resolvedOptions().timeZone;
// Collapse a run's `mode` (from WorkflowRunMode in api/enums.py) into a coarse
// channel. Telephony providers (twilio, plivo, telnyx, vonage, vobiz, cloudonix,
// ari, ...) are phone calls; webrtc/smallwebrtc are browser web calls; textchat
// is a text conversation. Anything unknown falls back to "phone".
const WEB_CALL_MODES = new Set(['webrtc', 'smallwebrtc']);
const TEXT_CHAT_MODES = new Set(['textchat']);
const getCallChannel = (mode?: string | null): 'phone' | 'web' | 'chat' => {
if (mode && TEXT_CHAT_MODES.has(mode)) return 'chat';
if (mode && WEB_CALL_MODES.has(mode)) return 'web';
return 'phone';
};
// Render the call's channel (mode) and direction (call_type) as two compact
// icons in a single cell, with a tooltip spelling out the full label. The
// channel icon shows medium/how (phone / web / chat); the colored arrow shows
// direction (inbound = incoming/emerald, outbound = outgoing/blue).
const CallTypeCell = ({ mode, callType }: { mode?: string | null; callType?: string | null }) => {
if (!mode && !callType) {
return <span className="text-sm text-muted-foreground">-</span>;
}
const channel = getCallChannel(mode);
const ChannelIcon = channel === 'chat' ? MessageSquare : channel === 'web' ? Globe : Phone;
const channelLabel = channel === 'chat' ? 'Text chat' : channel === 'web' ? 'Web call' : 'Phone call';
const isInbound = callType === 'inbound';
const DirectionIcon = isInbound ? ArrowDownLeft : ArrowUpRight;
const directionLabel = isInbound ? 'Inbound' : 'Outbound';
return (
<Tooltip>
<TooltipTrigger asChild>
<span className="inline-flex items-center gap-1">
<ChannelIcon className="h-4 w-4 text-muted-foreground" />
<DirectionIcon
className={`h-3.5 w-3.5 ${isInbound ? 'text-emerald-600' : 'text-blue-600'}`}
/>
</span>
</TooltipTrigger>
<TooltipContent sideOffset={4}>
{directionLabel} · {channelLabel}
</TooltipContent>
</Tooltip>
);
};
export default function UsagePage() {
const router = useRouter();
const searchParams = useSearchParams();
@ -534,13 +582,7 @@ export default function UsagePage() {
</TableCell>
<TableCell>{run.workflow_name || 'Unknown'}</TableCell>
<TableCell>
{run.call_type ? (
<Badge variant={run.call_type === 'inbound' ? "secondary" : "default"}>
{run.call_type === 'inbound' ? 'Inbound' : 'Outbound'}
</Badge>
) : (
<span className="text-sm text-muted-foreground">-</span>
)}
<CallTypeCell mode={run.mode} callType={run.call_type} />
</TableCell>
<TableCell className="text-sm">
{(run.call_type === 'inbound'

78
ui/src/app/workflow/[workflowId]/components/WorkflowEditorHeader.tsx
View file

@ -385,45 +385,6 @@ export const WorkflowEditorHeader = ({
</Popover>
)}
<Button
variant="outline"
className="flex items-center gap-2 bg-transparent border-[#3a3a3a] hover:bg-[#2a2a2a] text-white"
onClick={onTestAgentClick}
>
<Bot className="w-4 h-4" />
Test Agent
</Button>
{!isViewingHistoricalVersion && (
<Button
variant="outline"
className="flex items-center gap-2 bg-transparent border-[#3a3a3a] hover:bg-[#2a2a2a] text-white"
disabled={isCallDisabled}
onClick={onPhoneCallClick}
>
<Phone className="w-4 h-4" />
Phone Call
</Button>
)}
{/* Save button (only shown when editing the draft) */}
{!isViewingHistoricalVersion && (
<Button
onClick={handleSave}
disabled={!isDirty || savingWorkflow}
className="bg-teal-600 hover:bg-teal-700 text-white px-4"
>
{savingWorkflow ? (
<>
<LoaderCircle className="w-4 h-4 mr-2 animate-spin" />
Saving...
</>
) : (
"Save"
)}
</Button>
)}
{/* Publish button (only when on draft with no unsaved changes) */}
{!isViewingHistoricalVersion && hasDraft && (
<Button
@ -446,6 +407,45 @@ export const WorkflowEditorHeader = ({
</Button>
)}
{!isViewingHistoricalVersion && (
<Button
variant="outline"
className="flex items-center gap-2 bg-transparent border-[#3a3a3a] hover:bg-[#2a2a2a] text-white"
disabled={isCallDisabled}
onClick={onPhoneCallClick}
>
<Phone className="w-4 h-4" />
Phone Call
</Button>
)}
<Button
variant="outline"
className="flex items-center gap-2 bg-transparent border-[#3a3a3a] hover:bg-[#2a2a2a] text-white"
onClick={onTestAgentClick}
>
<Bot className="w-4 h-4" />
Test Agent
</Button>
{/* Save button (only shown when editing the draft) */}
{!isViewingHistoricalVersion && (
<Button
onClick={handleSave}
disabled={!isDirty || savingWorkflow}
className="bg-teal-600 hover:bg-teal-700 text-white px-4"
>
{savingWorkflow ? (
<>
<LoaderCircle className="w-4 h-4 mr-2 animate-spin" />
Saving...
</>
) : (
"Save"
)}
</Button>
)}
{/* More options dropdown */}
<DropdownMenu>
<DropdownMenuTrigger asChild>

113
ui/src/app/workflow/[workflowId]/components/WorkflowTesterPanel.tsx
View file

@ -74,6 +74,12 @@ type TextChatSession = Omit<WorkflowRunTextSessionResponse, "session_data" | "ch
checkpoint: TextChatCheckpoint;
};
interface TextChatToolEvent {
kind: "start" | "result";
functionName: string;
resultText?: string;
}
function toTextChatSession(response: WorkflowRunTextSessionResponse): TextChatSession {
return {
...response,
@ -182,6 +188,66 @@ function TypingBubble() {
);
}
function stringifyToolResult(result: unknown) {
if (result == null) return "No result";
if (typeof result === "string") return result;
try {
return JSON.stringify(result);
} catch {
return String(result);
}
}
function extractToolEvents(events: Array<Record<string, unknown>>): TextChatToolEvent[] {
return events.reduce<TextChatToolEvent[]>((acc, event) => {
const eventType = event.type;
const payload = event.payload;
if (!payload || typeof payload !== "object") {
return acc;
}
const typedPayload = payload as Record<string, unknown>;
const functionName = typeof typedPayload.function_name === "string"
? typedPayload.function_name
: "tool";
if (eventType === "tool_call_started") {
acc.push({ kind: "start", functionName });
return acc;
}
if (eventType === "tool_call_result") {
acc.push({
kind: "result",
functionName,
resultText: stringifyToolResult(typedPayload.result),
});
return acc;
}
return acc;
}, []);
}
function ToolEventBubble({ event }: { event: TextChatToolEvent }) {
return (
<div className="flex justify-start">
<div className="max-w-[85%] rounded-2xl rounded-bl-md border border-border/70 bg-background px-3.5 py-2 text-sm leading-6 text-foreground">
<div className="flex items-center gap-2">
<Badge variant="outline" className="h-5 px-1.5 text-[10px] uppercase tracking-[0.14em]">
{event.kind === "start" ? "Tool" : "Result"}
</Badge>
<span className="font-mono text-xs text-muted-foreground">
{event.kind === "start"
? `${event.functionName}()`
: `${event.functionName} -> ${event.resultText ?? "No result"}`}
</span>
</div>
</div>
</div>
);
}
function EmbeddedVoiceTester({
workflowId,
workflowRunId,
@ -357,14 +423,17 @@ function ManualTextChat({
initialContextVariables,
disabled,
disabledReason,
onActiveChange,
}: {
workflowId: number;
ready: boolean;
initialContextVariables?: Record<string, string>;
disabled: boolean;
disabledReason: string | null;
onActiveChange?: (active: boolean) => void;
}) {
const [session, setSession] = useState<TextChatSession | null>(null);
const [started, setStarted] = useState(false);
const [draft, setDraft] = useState("");
const [creatingSession, setCreatingSession] = useState(false);
const [sendingMessage, setSendingMessage] = useState(false);
@ -403,11 +472,15 @@ function ManualTextChat({
}, [disabled, initialContextVariables, workflowId]);
useEffect(() => {
if (creatingSession || session || !ready || disabled) {
if (!started || creatingSession || session || !ready || disabled) {
return;
}
void createSession();
}, [createSession, creatingSession, disabled, ready, session]);
}, [createSession, creatingSession, disabled, ready, session, started]);
useEffect(() => {
onActiveChange?.(started);
}, [onActiveChange, started]);
const sendMessage = useCallback(async () => {
if (!session || !draft.trim() || disabled) return;
@ -460,6 +533,25 @@ function ManualTextChat({
const inputDisabled = disabled || !session;
if (!started && !session) {
return (
<div className="flex h-full min-h-0 flex-col gap-3">
{disabledReason ? <DisabledNotice reason={disabledReason} /> : null}
<EmptyState
icon={<MessageSquareText className="h-7 w-7" />}
title="Chat with this agent"
description="Test the agent over a text conversation. Send messages and see how it responds, with tool calls and rewind support."
action={
<Button onClick={() => setStarted(true)} disabled={disabled || !ready}>
<MessageSquareText className="h-4 w-4" />
Start Test
</Button>
}
/>
</div>
);
}
return (
<div className="flex min-h-0 flex-1 flex-col">
{disabledReason ? (
@ -484,11 +576,19 @@ function ManualTextChat({
</div>
) : (
<div className="space-y-3 py-1">
{turns.map((turn) => (
{turns.map((turn) => {
const toolEvents = extractToolEvents(turn.events);
return (
<div key={turn.id} className="group space-y-1.5">
{turn.user_message ? (
<MessageBubble role="user" text={turn.user_message.text} />
) : null}
{toolEvents.map((event, index) => (
<ToolEventBubble
key={`${turn.id}-${event.kind}-${event.functionName}-${index}`}
event={event}
/>
))}
{turn.assistant_message ? (
<MessageBubble role="agent" text={turn.assistant_message.text} />
) : turn.status === "failed" ? (
@ -510,7 +610,8 @@ function ManualTextChat({
</button>
</div>
</div>
))}
);
})}
{sendingMessage ? <TypingBubble /> : null}
<div ref={scrollEndRef} />
</div>
@ -634,6 +735,7 @@ export function WorkflowTesterPanel({
const [activeMode, setActiveMode] = useState<"audio" | "text">("audio");
const [chatMode, setChatMode] = useState<"manual" | "simulated">("manual");
const [chatSessionKey, setChatSessionKey] = useState(0);
const [chatActive, setChatActive] = useState(false);
const [voiceRunId, setVoiceRunId] = useState<number | null>(null);
const [creatingVoiceRun, setCreatingVoiceRun] = useState(false);
const [tokenReady, setTokenReady] = useState(false);
@ -788,7 +890,7 @@ export function WorkflowTesterPanel({
<div className="flex h-full min-h-0 flex-col gap-3">
<div className="flex items-center justify-between gap-2">
<ChatModeToggle value={chatMode} onChange={setChatMode} />
{chatMode === "manual" ? (
{chatMode === "manual" && chatActive ? (
<Button
variant="ghost"
size="sm"
@ -810,6 +912,7 @@ export function WorkflowTesterPanel({
initialContextVariables={initialContextVariables}
disabled={testerBlocked}
disabledReason={effectiveDisabledReason}
onActiveChange={setChatActive}
/>
) : (
<AiSimulatorPlaceholder disabledReason={effectiveDisabledReason} />

136
ui/src/app/workflow/[workflowId]/run/[runId]/page.tsx
View file

@ -27,6 +27,7 @@ import { downloadFile } from '@/lib/files';
import { getRandomId } from '@/lib/utils';
interface WorkflowRunResponse {
mode: string;
is_completed: boolean;
transcript_url: string | null;
recording_url: string | null;
@ -183,6 +184,7 @@ export default function WorkflowRunPage() {
});
setIsLoading(false);
const runData = {
mode: response.data?.mode ?? '',
is_completed: response.data?.is_completed ?? false,
transcript_url: response.data?.transcript_url ?? null,
recording_url: response.data?.recording_url ?? null,
@ -223,6 +225,8 @@ export default function WorkflowRunPage() {
};
let returnValue = null;
const isTextChatRun = workflowRun?.mode === WORKFLOW_RUN_MODES.TEXTCHAT;
const showHistoricalRunView = Boolean(workflowRun?.is_completed || isTextChatRun);
if (isLoading) {
returnValue = (
@ -246,7 +250,7 @@ export default function WorkflowRunPage() {
</div>
);
}
else if (workflowRun?.is_completed) {
else if (showHistoricalRunView) {
returnValue = (
<div className={`flex ${RUN_SHELL_HEIGHT_CLASS} min-h-0 w-full overflow-hidden bg-background`}>
<div className="min-w-0 flex-1 overflow-y-auto">
@ -254,27 +258,35 @@ export default function WorkflowRunPage() {
<Card className="border-border">
<CardHeader className="flex flex-row items-center justify-between">
<div className="flex items-center gap-4">
<CardTitle className="text-2xl">Agent Run Completed</CardTitle>
<div className="h-8 w-8 bg-emerald-500/20 rounded-full flex items-center justify-center">
<svg className="h-5 w-5 text-emerald-500" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth="2" d="M5 13l4 4L19 7" />
</svg>
<CardTitle className="text-2xl">
{isTextChatRun ? 'Text Chat Session' : 'Agent Run Completed'}
</CardTitle>
<div className={`h-8 w-8 rounded-full flex items-center justify-center ${isTextChatRun ? 'bg-sky-500/15' : 'bg-emerald-500/20'}`}>
{isTextChatRun ? (
<FileText className="h-5 w-5 text-sky-500" />
) : (
<svg className="h-5 w-5 text-emerald-500" fill="none" stroke="currentColor" viewBox="0 0 24 24">
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth="2" d="M5 13l4 4L19 7" />
</svg>
)}
</div>
</div>
<div className="flex items-center gap-2">
<Button
onClick={handleTestAgain}
disabled={startingCall}
variant="outline"
className="gap-2"
>
{startingCall ? (
<LoaderCircle className="h-4 w-4 animate-spin" />
) : (
<Phone className="h-4 w-4" />
)}
{startingCall ? 'Starting...' : 'Test Again'}
</Button>
{!isTextChatRun && (
<Button
onClick={handleTestAgain}
disabled={startingCall}
variant="outline"
className="gap-2"
>
{startingCall ? (
<LoaderCircle className="h-4 w-4 animate-spin" />
) : (
<Phone className="h-4 w-4" />
)}
{startingCall ? 'Starting...' : 'Test Again'}
</Button>
)}
<Link href={`/workflow/${params.workflowId}`}>
<Button
ref={customizeButtonRef}
@ -294,41 +306,49 @@ export default function WorkflowRunPage() {
</div>
</CardHeader>
<CardContent>
<p className="text-muted-foreground mb-8">Your voice agent run has been completed successfully. You can preview or download the transcript and recording.</p>
<p className="text-muted-foreground mb-8">
{isTextChatRun
? 'Review the conversation history, metrics, and context captured for this text session.'
: 'Your voice agent run has been completed successfully. You can preview or download the transcript and recording.'}
</p>
<div className="flex flex-wrap gap-4">
<div className="flex items-center gap-2">
<span className="text-sm text-muted-foreground">Preview:</span>
<MediaPreviewButton
recordingUrl={workflowRun?.recording_url}
transcriptUrl={workflowRun?.transcript_url}
runId={Number(params.runId)}
onOpenPreview={openPreview}
/>
</div>
<div className="flex items-center gap-2 border-l border-border pl-4">
<span className="text-sm text-muted-foreground">Download:</span>
<Button
onClick={() => downloadFile(workflowRun?.transcript_url)}
disabled={!workflowRun?.transcript_url || !auth.isAuthenticated}
size="sm"
className="gap-2"
>
<FileText className="h-4 w-4" />
Transcript
</Button>
<Button
onClick={() => downloadFile(workflowRun?.recording_url)}
disabled={!workflowRun?.recording_url || !auth.isAuthenticated}
size="sm"
className="gap-2"
>
<Video className="h-4 w-4" />
Recording
</Button>
</div>
{!isTextChatRun && (
<>
<div className="flex items-center gap-2">
<span className="text-sm text-muted-foreground">Preview:</span>
<MediaPreviewButton
recordingUrl={workflowRun?.recording_url}
transcriptUrl={workflowRun?.transcript_url}
runId={Number(params.runId)}
onOpenPreview={openPreview}
/>
</div>
<div className="flex items-center gap-2 border-l border-border pl-4">
<span className="text-sm text-muted-foreground">Download:</span>
<Button
onClick={() => downloadFile(workflowRun?.transcript_url ?? null)}
disabled={!workflowRun?.transcript_url || !auth.isAuthenticated}
size="sm"
className="gap-2"
>
<FileText className="h-4 w-4" />
Transcript
</Button>
<Button
onClick={() => downloadFile(workflowRun?.recording_url ?? null)}
disabled={!workflowRun?.recording_url || !auth.isAuthenticated}
size="sm"
className="gap-2"
>
<Video className="h-4 w-4" />
Recording
</Button>
</div>
</>
)}
{workflowRun?.gathered_context?.trace_url && (
<div className="flex items-center gap-2 border-l border-border pl-4">
<div className={`flex items-center gap-2 ${isTextChatRun ? '' : 'border-l border-border pl-4'}`}>
<span className="text-sm text-muted-foreground">Trace:</span>
<Button
asChild
@ -352,19 +372,19 @@ export default function WorkflowRunPage() {
</Card>
<RunMetricsSection
costInfo={workflowRun?.cost_info}
logs={workflowRun?.logs}
gatheredContext={workflowRun?.gathered_context}
costInfo={workflowRun?.cost_info ?? null}
logs={workflowRun?.logs ?? null}
gatheredContext={workflowRun?.gathered_context ?? null}
/>
<div className="grid gap-6 md:grid-cols-2">
<ContextDisplay
title="Initial Context"
context={workflowRun?.initial_context}
context={workflowRun?.initial_context ?? null}
/>
<ContextDisplay
title="Gathered Context"
context={workflowRun?.gathered_context}
context={workflowRun?.gathered_context ?? null}
/>
</div>
@ -379,7 +399,7 @@ export default function WorkflowRunPage() {
<div className="h-full min-h-0 w-[420px] shrink-0 border-l border-border bg-background p-5">
<TranscriptRailFrame className="h-full">
<RealtimeFeedback mode="historical" logs={workflowRun?.logs} />
<RealtimeFeedback mode="historical" logs={workflowRun?.logs ?? null} />
</TranscriptRailFrame>
</div>
</div>
@ -411,7 +431,7 @@ export default function WorkflowRunPage() {
{dialog}
{/* Onboarding Tooltip for Customize Workflow */}
{workflowRun?.is_completed && (
{showHistoricalRunView && (
<OnboardingTooltip
title='Customize Your Workflow'
targetRef={customizeButtonRef}

4
ui/src/client/types.gen.ts
View file

@ -4675,6 +4675,10 @@ export type WorkflowRunUsageResponse = {
* Call Type
*/
call_type?: string | null;
/**
* Mode
*/
mode?: string | null;
/**
* Disposition
*/