mirror of
https://github.com/Monadical-SAS/reflector.git
synced 2025-12-20 20:29:06 +00:00
Compare commits
30 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f0ee7b531a | |||
| 37a454f283 | |||
| 964cd78bb6 | |||
| 5f458aa4a7 | |||
| 5f7dfadabd | |||
| 0bc971ba96 | |||
|
|
c62e3c0753 | ||
|
|
16284e1ac3 | ||
|
|
443982617d | ||
|
|
23023b3cdb | ||
| 90c3ecc9c3 | |||
| d7f140b7d1 | |||
| a47a5f5781 | |||
| 0eba147018 | |||
| 18a27f7b45 | |||
| 32a049c134 | |||
| 91650ec65f | |||
|
|
61f0e29d4c | ||
|
|
ec17ed7b58 | ||
|
|
00549f153a | ||
| 3ad78be762 | |||
| d3a5cd12d2 | |||
| af921ce927 | |||
|
|
bd5df1ce2e | ||
| c8024484b3 | |||
| 28f87c09dc | |||
| dabf7251db | |||
|
|
b51b7aa917 | ||
|
|
a8983b4e7e | ||
|
|
fe47c46489 |
90
.github/workflows/deploy.yml
vendored
90
.github/workflows/deploy.yml
vendored
@@ -1,90 +0,0 @@
|
||||
name: Build container/push to container registry
|
||||
|
||||
on: [workflow_dispatch]
|
||||
|
||||
env:
|
||||
# 950402358378.dkr.ecr.us-east-1.amazonaws.com/reflector
|
||||
AWS_REGION: us-east-1
|
||||
ECR_REPOSITORY: reflector
|
||||
|
||||
jobs:
|
||||
build:
|
||||
strategy:
|
||||
matrix:
|
||||
include:
|
||||
- platform: linux/amd64
|
||||
runner: linux-amd64
|
||||
arch: amd64
|
||||
- platform: linux/arm64
|
||||
runner: linux-arm64
|
||||
arch: arm64
|
||||
|
||||
runs-on: ${{ matrix.runner }}
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
outputs:
|
||||
registry: ${{ steps.login-ecr.outputs.registry }}
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Configure AWS credentials
|
||||
uses: aws-actions/configure-aws-credentials@v4
|
||||
with:
|
||||
aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
|
||||
aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
|
||||
aws-region: ${{ env.AWS_REGION }}
|
||||
|
||||
- name: Login to Amazon ECR
|
||||
id: login-ecr
|
||||
uses: aws-actions/amazon-ecr-login@v2
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3
|
||||
|
||||
- name: Build and push ${{ matrix.arch }}
|
||||
uses: docker/build-push-action@v5
|
||||
with:
|
||||
context: server
|
||||
platforms: ${{ matrix.platform }}
|
||||
push: true
|
||||
tags: ${{ steps.login-ecr.outputs.registry }}/${{ env.ECR_REPOSITORY }}:latest-${{ matrix.arch }}
|
||||
cache-from: type=gha,scope=${{ matrix.arch }}
|
||||
cache-to: type=gha,mode=max,scope=${{ matrix.arch }}
|
||||
github-token: ${{ secrets.GHA_CACHE_TOKEN }}
|
||||
provenance: false
|
||||
|
||||
create-manifest:
|
||||
runs-on: ubuntu-latest
|
||||
needs: [build]
|
||||
|
||||
permissions:
|
||||
deployments: write
|
||||
contents: read
|
||||
|
||||
steps:
|
||||
- name: Configure AWS credentials
|
||||
uses: aws-actions/configure-aws-credentials@v4
|
||||
with:
|
||||
aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
|
||||
aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
|
||||
aws-region: ${{ env.AWS_REGION }}
|
||||
|
||||
- name: Login to Amazon ECR
|
||||
uses: aws-actions/amazon-ecr-login@v2
|
||||
|
||||
- name: Create and push multi-arch manifest
|
||||
run: |
|
||||
# Get the registry URL (since we can't easily access job outputs in matrix)
|
||||
ECR_REGISTRY=$(aws ecr describe-registry --query 'registryId' --output text).dkr.ecr.${{ env.AWS_REGION }}.amazonaws.com
|
||||
|
||||
docker manifest create \
|
||||
$ECR_REGISTRY/${{ env.ECR_REPOSITORY }}:latest \
|
||||
$ECR_REGISTRY/${{ env.ECR_REPOSITORY }}:latest-amd64 \
|
||||
$ECR_REGISTRY/${{ env.ECR_REPOSITORY }}:latest-arm64
|
||||
|
||||
docker manifest push $ECR_REGISTRY/${{ env.ECR_REPOSITORY }}:latest
|
||||
|
||||
echo "✅ Multi-arch manifest pushed: $ECR_REGISTRY/${{ env.ECR_REPOSITORY }}:latest"
|
||||
@@ -1,35 +1,31 @@
|
||||
name: Build and Push Frontend Docker Image
|
||||
name: Build and Push Backend Docker Image (Docker Hub)
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
paths:
|
||||
- 'www/**'
|
||||
- '.github/workflows/docker-frontend.yml'
|
||||
tags:
|
||||
- "v*"
|
||||
workflow_dispatch:
|
||||
|
||||
env:
|
||||
REGISTRY: ghcr.io
|
||||
IMAGE_NAME: ${{ github.repository }}-frontend
|
||||
REGISTRY: docker.io
|
||||
IMAGE_NAME: monadicalsas/reflector-backend
|
||||
|
||||
jobs:
|
||||
build-and-push:
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
packages: write
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Log in to GitHub Container Registry
|
||||
- name: Log in to Docker Hub
|
||||
uses: docker/login-action@v3
|
||||
with:
|
||||
registry: ${{ env.REGISTRY }}
|
||||
username: ${{ github.actor }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
username: monadicalsas
|
||||
password: ${{ secrets.DOCKERHUB_TOKEN }}
|
||||
|
||||
- name: Extract metadata
|
||||
id: meta
|
||||
@@ -38,7 +34,7 @@ jobs:
|
||||
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
|
||||
tags: |
|
||||
type=ref,event=branch
|
||||
type=sha,prefix={{branch}}-
|
||||
type=ref,event=tag
|
||||
type=raw,value=latest,enable={{is_default_branch}}
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
@@ -47,11 +43,11 @@ jobs:
|
||||
- name: Build and push Docker image
|
||||
uses: docker/build-push-action@v5
|
||||
with:
|
||||
context: ./www
|
||||
file: ./www/Dockerfile
|
||||
context: ./server
|
||||
file: ./server/Dockerfile
|
||||
push: true
|
||||
tags: ${{ steps.meta.outputs.tags }}
|
||||
labels: ${{ steps.meta.outputs.labels }}
|
||||
cache-from: type=gha
|
||||
cache-to: type=gha,mode=max
|
||||
platforms: linux/amd64,linux/arm64
|
||||
platforms: linux/amd64,linux/arm64
|
||||
70
.github/workflows/dockerhub-frontend.yml
vendored
Normal file
70
.github/workflows/dockerhub-frontend.yml
vendored
Normal file
@@ -0,0 +1,70 @@
|
||||
name: Build and Push Frontend Docker Image
|
||||
|
||||
on:
|
||||
push:
|
||||
tags:
|
||||
- "v*"
|
||||
workflow_dispatch:
|
||||
|
||||
env:
|
||||
REGISTRY: docker.io
|
||||
IMAGE_NAME: monadicalsas/reflector-frontend
|
||||
|
||||
jobs:
|
||||
build-and-push:
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Log in to Docker Hub
|
||||
uses: docker/login-action@v3
|
||||
with:
|
||||
registry: ${{ env.REGISTRY }}
|
||||
username: monadicalsas
|
||||
password: ${{ secrets.DOCKERHUB_TOKEN }}
|
||||
|
||||
- name: Extract metadata
|
||||
id: meta
|
||||
uses: docker/metadata-action@v5
|
||||
with:
|
||||
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
|
||||
tags: |
|
||||
type=ref,event=branch
|
||||
type=ref,event=tag
|
||||
type=raw,value=latest,enable={{is_default_branch}}
|
||||
github-token: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3
|
||||
|
||||
- name: Build and push Docker image
|
||||
uses: docker/build-push-action@v5
|
||||
with:
|
||||
context: ./www
|
||||
file: ./www/Dockerfile
|
||||
push: true
|
||||
tags: ${{ steps.meta.outputs.tags }}
|
||||
labels: ${{ steps.meta.outputs.labels }}
|
||||
cache-from: type=gha
|
||||
cache-to: type=gha,mode=max
|
||||
platforms: linux/amd64,linux/arm64
|
||||
|
||||
deploy:
|
||||
needs: build-and-push
|
||||
runs-on: ubuntu-latest
|
||||
if: success()
|
||||
strategy:
|
||||
matrix:
|
||||
environment: [reflector-monadical, reflector-media]
|
||||
environment: ${{ matrix.environment }}
|
||||
steps:
|
||||
- name: Trigger Coolify deployment
|
||||
run: |
|
||||
curl -X POST "${{ secrets.COOLIFY_WEBHOOK_URL }}" \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Authorization: Bearer ${{ secrets.COOLIFY_WEBHOOK_TOKEN }}" \
|
||||
-f || (echo "Failed to trigger Coolify deployment for ${{ matrix.environment }}" && exit 1)
|
||||
1
.gitignore
vendored
1
.gitignore
vendored
@@ -18,3 +18,4 @@ CLAUDE.local.md
|
||||
www/.env.development
|
||||
www/.env.production
|
||||
.playwright-mcp
|
||||
.secrets
|
||||
|
||||
24
.secrets.example
Normal file
24
.secrets.example
Normal file
@@ -0,0 +1,24 @@
|
||||
# Example secrets file for GitHub Actions workflows
|
||||
# Copy this to .secrets and fill in your values
|
||||
# These secrets should be configured in GitHub repository settings:
|
||||
# Settings > Secrets and variables > Actions
|
||||
|
||||
# DockerHub Configuration (required for frontend and backend deployment)
|
||||
# Create a Docker Hub access token at https://hub.docker.com/settings/security
|
||||
# Username: monadicalsas
|
||||
DOCKERHUB_TOKEN=your-dockerhub-access-token
|
||||
|
||||
# GitHub Token (required for frontend and backend deployment)
|
||||
# Used by docker/metadata-action for extracting image metadata
|
||||
# Can use the default GITHUB_TOKEN or create a personal access token
|
||||
GITHUB_TOKEN=your-github-token-or-use-default-GITHUB_TOKEN
|
||||
|
||||
# Coolify Deployment Webhook (required for frontend deployment)
|
||||
# Used to trigger automatic deployment after image push
|
||||
# Configure these secrets in GitHub Environments:
|
||||
# Each environment should have:
|
||||
# - COOLIFY_WEBHOOK_URL: The webhook URL for that specific deployment
|
||||
# - COOLIFY_WEBHOOK_TOKEN: The webhook token (can be the same for both if using same token)
|
||||
|
||||
# Optional: GitHub Actions Cache Token (for local testing with act)
|
||||
GHA_CACHE_TOKEN=your-github-token-or-empty
|
||||
67
CHANGELOG.md
67
CHANGELOG.md
@@ -1,5 +1,72 @@
|
||||
# Changelog
|
||||
|
||||
## [0.24.0](https://github.com/Monadical-SAS/reflector/compare/v0.23.2...v0.24.0) (2025-12-18)
|
||||
|
||||
|
||||
### Features
|
||||
|
||||
* identify action items ([#790](https://github.com/Monadical-SAS/reflector/issues/790)) ([964cd78](https://github.com/Monadical-SAS/reflector/commit/964cd78bb699d83d012ae4b8c96565df25b90a5d))
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* automatically reprocess daily recordings ([#797](https://github.com/Monadical-SAS/reflector/issues/797)) ([5f458aa](https://github.com/Monadical-SAS/reflector/commit/5f458aa4a7ec3d00ca5ec49d62fcc8ad232b138e))
|
||||
* daily video optimisation ([#789](https://github.com/Monadical-SAS/reflector/issues/789)) ([16284e1](https://github.com/Monadical-SAS/reflector/commit/16284e1ac3faede2b74f0d91b50c0b5612af2c35))
|
||||
* main menu login ([#800](https://github.com/Monadical-SAS/reflector/issues/800)) ([0bc971b](https://github.com/Monadical-SAS/reflector/commit/0bc971ba966a52d719c8c240b47dc7b3bdea4391))
|
||||
* retry on workflow timeout ([#798](https://github.com/Monadical-SAS/reflector/issues/798)) ([5f7dfad](https://github.com/Monadical-SAS/reflector/commit/5f7dfadabd3e8017406ad3720ba495a59963ee34))
|
||||
|
||||
## [0.23.2](https://github.com/Monadical-SAS/reflector/compare/v0.23.1...v0.23.2) (2025-12-11)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* build on push tags ([#785](https://github.com/Monadical-SAS/reflector/issues/785)) ([d7f140b](https://github.com/Monadical-SAS/reflector/commit/d7f140b7d1f4660d5da7a0da1357f68869e0b5cd))
|
||||
|
||||
## [0.23.1](https://github.com/Monadical-SAS/reflector/compare/v0.23.0...v0.23.1) (2025-12-11)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* populate room_name in transcript GET endpoint ([#783](https://github.com/Monadical-SAS/reflector/issues/783)) ([0eba147](https://github.com/Monadical-SAS/reflector/commit/0eba1470181c7b9e0a79964a1ef28c09bcbdd9d7))
|
||||
|
||||
## [0.23.0](https://github.com/Monadical-SAS/reflector/compare/v0.22.4...v0.23.0) (2025-12-10)
|
||||
|
||||
|
||||
### Features
|
||||
|
||||
* dockerhub ci ([#772](https://github.com/Monadical-SAS/reflector/issues/772)) ([00549f1](https://github.com/Monadical-SAS/reflector/commit/00549f153ade922cf4cb6c5358a7d11a39c426d2))
|
||||
* llm retries ([#739](https://github.com/Monadical-SAS/reflector/issues/739)) ([61f0e29](https://github.com/Monadical-SAS/reflector/commit/61f0e29d4c51eab54ee67af92141fbb171e8ccaa))
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* celery inspect bug sidestep in restart script ([#766](https://github.com/Monadical-SAS/reflector/issues/766)) ([ec17ed7](https://github.com/Monadical-SAS/reflector/commit/ec17ed7b587cf6ee143646baaee67a7c017044d4))
|
||||
* deploy frontend to coolify ([#779](https://github.com/Monadical-SAS/reflector/issues/779)) ([91650ec](https://github.com/Monadical-SAS/reflector/commit/91650ec65f65713faa7ee0dcfb75af427b7c4ba0))
|
||||
* hide rooms settings instead of disabling ([#763](https://github.com/Monadical-SAS/reflector/issues/763)) ([3ad78be](https://github.com/Monadical-SAS/reflector/commit/3ad78be7628c0d029296b301a0e87236c76b7598))
|
||||
* return participant emails from transcript endpoint ([#769](https://github.com/Monadical-SAS/reflector/issues/769)) ([d3a5cd1](https://github.com/Monadical-SAS/reflector/commit/d3a5cd12d2d0d9c32af2d5bd9322e030ef69b85d))
|
||||
|
||||
## [0.22.4](https://github.com/Monadical-SAS/reflector/compare/v0.22.3...v0.22.4) (2025-12-02)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* Multitrack mixdown optimisation 2 ([#764](https://github.com/Monadical-SAS/reflector/issues/764)) ([bd5df1c](https://github.com/Monadical-SAS/reflector/commit/bd5df1ce2ebf35d7f3413b295e56937a9a28ef7b))
|
||||
|
||||
## [0.22.3](https://github.com/Monadical-SAS/reflector/compare/v0.22.2...v0.22.3) (2025-12-02)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* align daily room settings ([#759](https://github.com/Monadical-SAS/reflector/issues/759)) ([28f87c0](https://github.com/Monadical-SAS/reflector/commit/28f87c09dc459846873d0dde65b03e3d7b2b9399))
|
||||
|
||||
## [0.22.2](https://github.com/Monadical-SAS/reflector/compare/v0.22.1...v0.22.2) (2025-12-02)
|
||||
|
||||
|
||||
### Bug Fixes
|
||||
|
||||
* daily auto refresh fix ([#755](https://github.com/Monadical-SAS/reflector/issues/755)) ([fe47c46](https://github.com/Monadical-SAS/reflector/commit/fe47c46489c5aa0cc538109f7559cc9accb35c01))
|
||||
* Skip mixdown for multitrack ([#760](https://github.com/Monadical-SAS/reflector/issues/760)) ([b51b7aa](https://github.com/Monadical-SAS/reflector/commit/b51b7aa9176c1a53ba57ad99f5e976c804a1e80c))
|
||||
|
||||
## [0.22.1](https://github.com/Monadical-SAS/reflector/compare/v0.22.0...v0.22.1) (2025-11-27)
|
||||
|
||||
|
||||
|
||||
@@ -3,10 +3,8 @@
|
||||
|
||||
services:
|
||||
web:
|
||||
build:
|
||||
context: ./www
|
||||
dockerfile: Dockerfile
|
||||
image: reflector-frontend:latest
|
||||
image: monadicalsas/reflector-frontend:latest
|
||||
pull_policy: always
|
||||
environment:
|
||||
- KV_URL=${KV_URL:-redis://redis:6379}
|
||||
- SITE_URL=${SITE_URL}
|
||||
@@ -36,4 +34,4 @@ services:
|
||||
- redis_data:/data
|
||||
|
||||
volumes:
|
||||
redis_data:
|
||||
redis_data:
|
||||
|
||||
26
server/migrations/versions/05f8688d6895_add_action_items.py
Normal file
26
server/migrations/versions/05f8688d6895_add_action_items.py
Normal file
@@ -0,0 +1,26 @@
|
||||
"""add_action_items
|
||||
|
||||
Revision ID: 05f8688d6895
|
||||
Revises: bbafedfa510c
|
||||
Create Date: 2025-12-12 11:57:50.209658
|
||||
|
||||
"""
|
||||
|
||||
from typing import Sequence, Union
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "05f8688d6895"
|
||||
down_revision: Union[str, None] = "bbafedfa510c"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column("transcript", sa.Column("action_items", sa.JSON(), nullable=True))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_column("transcript", "action_items")
|
||||
@@ -126,6 +126,7 @@ markers = [
|
||||
select = [
|
||||
"I", # isort - import sorting
|
||||
"F401", # unused imports
|
||||
"E402", # module level import not at top of file
|
||||
"PLC0415", # import-outside-top-level - detect inline imports
|
||||
]
|
||||
|
||||
|
||||
@@ -1,13 +1,19 @@
|
||||
import asyncio
|
||||
import functools
|
||||
from uuid import uuid4
|
||||
|
||||
from celery import current_task
|
||||
|
||||
from reflector.db import get_database
|
||||
from reflector.llm import llm_session_id
|
||||
|
||||
|
||||
def asynctask(f):
|
||||
@functools.wraps(f)
|
||||
def wrapper(*args, **kwargs):
|
||||
async def run_with_db():
|
||||
task_id = current_task.request.id if current_task else None
|
||||
llm_session_id.set(task_id or f"random-{uuid4().hex}")
|
||||
database = get_database()
|
||||
await database.connect()
|
||||
try:
|
||||
|
||||
@@ -18,6 +18,7 @@ from .requests import (
|
||||
|
||||
# Response models
|
||||
from .responses import (
|
||||
FinishedRecordingResponse,
|
||||
MeetingParticipant,
|
||||
MeetingParticipantsResponse,
|
||||
MeetingResponse,
|
||||
@@ -79,6 +80,7 @@ __all__ = [
|
||||
"MeetingParticipant",
|
||||
"MeetingResponse",
|
||||
"RecordingResponse",
|
||||
"FinishedRecordingResponse",
|
||||
"RecordingS3Info",
|
||||
"MeetingTokenResponse",
|
||||
"WebhookResponse",
|
||||
|
||||
@@ -40,6 +40,10 @@ class RoomProperties(BaseModel):
|
||||
)
|
||||
enable_chat: bool = Field(default=True, description="Enable in-meeting chat")
|
||||
enable_screenshare: bool = Field(default=True, description="Enable screen sharing")
|
||||
enable_knocking: bool = Field(
|
||||
default=False,
|
||||
description="Enable knocking for private rooms (allows participants to request access)",
|
||||
)
|
||||
start_video_off: bool = Field(
|
||||
default=False, description="Start with video off for all participants"
|
||||
)
|
||||
|
||||
@@ -121,7 +121,10 @@ class RecordingS3Info(BaseModel):
|
||||
|
||||
class RecordingResponse(BaseModel):
|
||||
"""
|
||||
Response from recording retrieval endpoint.
|
||||
Response from recording retrieval endpoint (network layer).
|
||||
|
||||
Duration may be None for recordings still being processed by Daily.
|
||||
Use FinishedRecordingResponse for recordings ready for processing.
|
||||
|
||||
Reference: https://docs.daily.co/reference/rest-api/recordings
|
||||
"""
|
||||
@@ -135,7 +138,9 @@ class RecordingResponse(BaseModel):
|
||||
max_participants: int | None = Field(
|
||||
None, description="Maximum participants during recording (may be missing)"
|
||||
)
|
||||
duration: int = Field(description="Recording duration in seconds")
|
||||
duration: int | None = Field(
|
||||
None, description="Recording duration in seconds (None if still processing)"
|
||||
)
|
||||
share_token: NonEmptyString | None = Field(
|
||||
None, description="Token for sharing recording"
|
||||
)
|
||||
@@ -149,6 +154,25 @@ class RecordingResponse(BaseModel):
|
||||
None, description="Meeting session identifier (may be missing)"
|
||||
)
|
||||
|
||||
def to_finished(self) -> "FinishedRecordingResponse | None":
|
||||
"""Convert to FinishedRecordingResponse if duration is available and status is finished."""
|
||||
if self.duration is None or self.status != "finished":
|
||||
return None
|
||||
return FinishedRecordingResponse(**self.model_dump())
|
||||
|
||||
|
||||
class FinishedRecordingResponse(RecordingResponse):
|
||||
"""
|
||||
Recording with confirmed duration - ready for processing.
|
||||
|
||||
This model guarantees duration is present and status is finished.
|
||||
"""
|
||||
|
||||
status: Literal["finished"] = Field(
|
||||
description="Recording status (always 'finished')"
|
||||
)
|
||||
duration: int = Field(description="Recording duration in seconds")
|
||||
|
||||
|
||||
class MeetingTokenResponse(BaseModel):
|
||||
"""
|
||||
|
||||
@@ -3,6 +3,7 @@ from typing import Literal
|
||||
|
||||
import sqlalchemy as sa
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy import or_
|
||||
|
||||
from reflector.db import get_database, metadata
|
||||
from reflector.utils import generate_uuid4
|
||||
@@ -79,5 +80,35 @@ class RecordingController:
|
||||
results = await get_database().fetch_all(query)
|
||||
return [Recording(**row) for row in results]
|
||||
|
||||
async def get_multitrack_needing_reprocessing(
|
||||
self, bucket_name: str
|
||||
) -> list[Recording]:
|
||||
"""
|
||||
Get multitrack recordings that need reprocessing:
|
||||
- Have track_keys (multitrack)
|
||||
- Either have no transcript OR transcript has error status
|
||||
|
||||
This is more efficient than fetching all recordings and filtering in Python.
|
||||
"""
|
||||
from reflector.db.transcripts import (
|
||||
transcripts, # noqa: PLC0415 cyclic import
|
||||
)
|
||||
|
||||
query = (
|
||||
recordings.select()
|
||||
.outerjoin(transcripts, recordings.c.id == transcripts.c.recording_id)
|
||||
.where(
|
||||
recordings.c.bucket_name == bucket_name,
|
||||
recordings.c.track_keys.isnot(None),
|
||||
or_(
|
||||
transcripts.c.id.is_(None),
|
||||
transcripts.c.status == "error",
|
||||
),
|
||||
)
|
||||
)
|
||||
results = await get_database().fetch_all(query)
|
||||
recordings_list = [Recording(**row) for row in results]
|
||||
return [r for r in recordings_list if r.is_multitrack]
|
||||
|
||||
|
||||
recordings_controller = RecordingController()
|
||||
|
||||
@@ -44,6 +44,7 @@ transcripts = sqlalchemy.Table(
|
||||
sqlalchemy.Column("title", sqlalchemy.String),
|
||||
sqlalchemy.Column("short_summary", sqlalchemy.String),
|
||||
sqlalchemy.Column("long_summary", sqlalchemy.String),
|
||||
sqlalchemy.Column("action_items", sqlalchemy.JSON),
|
||||
sqlalchemy.Column("topics", sqlalchemy.JSON),
|
||||
sqlalchemy.Column("events", sqlalchemy.JSON),
|
||||
sqlalchemy.Column("participants", sqlalchemy.JSON),
|
||||
@@ -164,6 +165,10 @@ class TranscriptFinalLongSummary(BaseModel):
|
||||
long_summary: str
|
||||
|
||||
|
||||
class TranscriptActionItems(BaseModel):
|
||||
action_items: dict
|
||||
|
||||
|
||||
class TranscriptFinalTitle(BaseModel):
|
||||
title: str
|
||||
|
||||
@@ -204,6 +209,7 @@ class Transcript(BaseModel):
|
||||
locked: bool = False
|
||||
short_summary: str | None = None
|
||||
long_summary: str | None = None
|
||||
action_items: dict | None = None
|
||||
topics: list[TranscriptTopic] = []
|
||||
events: list[TranscriptEvent] = []
|
||||
participants: list[TranscriptParticipant] | None = []
|
||||
@@ -368,7 +374,12 @@ class TranscriptController:
|
||||
room_id: str | None = None,
|
||||
search_term: str | None = None,
|
||||
return_query: bool = False,
|
||||
exclude_columns: list[str] = ["topics", "events", "participants"],
|
||||
exclude_columns: list[str] = [
|
||||
"topics",
|
||||
"events",
|
||||
"participants",
|
||||
"action_items",
|
||||
],
|
||||
) -> list[Transcript]:
|
||||
"""
|
||||
Get all transcripts
|
||||
|
||||
@@ -88,5 +88,11 @@ class UserController:
|
||||
results = await get_database().fetch_all(query)
|
||||
return [User(**r) for r in results]
|
||||
|
||||
@staticmethod
|
||||
async def get_by_ids(user_ids: list[NonEmptyString]) -> dict[str, User]:
|
||||
query = users.select().where(users.c.id.in_(user_ids))
|
||||
results = await get_database().fetch_all(query)
|
||||
return {user.id: User(**user) for user in results}
|
||||
|
||||
|
||||
user_controller = UserController()
|
||||
|
||||
@@ -1,14 +1,32 @@
|
||||
import logging
|
||||
from typing import Type, TypeVar
|
||||
from contextvars import ContextVar
|
||||
from typing import Generic, Type, TypeVar
|
||||
from uuid import uuid4
|
||||
|
||||
from llama_index.core import Settings
|
||||
from llama_index.core.output_parsers import PydanticOutputParser
|
||||
from llama_index.core.program import LLMTextCompletionProgram
|
||||
from llama_index.core.response_synthesizers import TreeSummarize
|
||||
from llama_index.core.workflow import (
|
||||
Context,
|
||||
Event,
|
||||
StartEvent,
|
||||
StopEvent,
|
||||
Workflow,
|
||||
step,
|
||||
)
|
||||
from llama_index.llms.openai_like import OpenAILike
|
||||
from pydantic import BaseModel, ValidationError
|
||||
from workflows.errors import WorkflowTimeoutError
|
||||
|
||||
from reflector.utils.retry import retry
|
||||
|
||||
T = TypeVar("T", bound=BaseModel)
|
||||
OutputT = TypeVar("OutputT", bound=BaseModel)
|
||||
|
||||
# Session ID for LiteLLM request grouping - set per processing run
|
||||
llm_session_id: ContextVar[str | None] = ContextVar("llm_session_id", default=None)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
STRUCTURED_RESPONSE_PROMPT_TEMPLATE = """
|
||||
Based on the following analysis, provide the information in the requested JSON format:
|
||||
@@ -20,6 +38,158 @@ Analysis:
|
||||
"""
|
||||
|
||||
|
||||
class LLMParseError(Exception):
|
||||
"""Raised when LLM output cannot be parsed after retries."""
|
||||
|
||||
def __init__(self, output_cls: Type[BaseModel], error_msg: str, attempts: int):
|
||||
self.output_cls = output_cls
|
||||
self.error_msg = error_msg
|
||||
self.attempts = attempts
|
||||
super().__init__(
|
||||
f"Failed to parse {output_cls.__name__} after {attempts} attempts: {error_msg}"
|
||||
)
|
||||
|
||||
|
||||
class ExtractionDone(Event):
|
||||
"""Event emitted when LLM JSON formatting completes."""
|
||||
|
||||
output: str
|
||||
|
||||
|
||||
class ValidationErrorEvent(Event):
|
||||
"""Event emitted when validation fails."""
|
||||
|
||||
error: str
|
||||
wrong_output: str
|
||||
|
||||
|
||||
class StructuredOutputWorkflow(Workflow, Generic[OutputT]):
|
||||
"""Workflow for structured output extraction with validation retry.
|
||||
|
||||
This workflow handles parse/validation retries only. Network error retries
|
||||
are handled internally by Settings.llm (OpenAILike max_retries=3).
|
||||
The caller should NOT wrap this workflow in additional retry logic.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
output_cls: Type[OutputT],
|
||||
max_retries: int = 3,
|
||||
**kwargs,
|
||||
):
|
||||
super().__init__(**kwargs)
|
||||
self.output_cls: Type[OutputT] = output_cls
|
||||
self.max_retries = max_retries
|
||||
self.output_parser = PydanticOutputParser(output_cls)
|
||||
|
||||
@step
|
||||
async def extract(
|
||||
self, ctx: Context, ev: StartEvent | ValidationErrorEvent
|
||||
) -> StopEvent | ExtractionDone:
|
||||
"""Extract structured data from text using two-step LLM process.
|
||||
|
||||
Step 1 (first call only): TreeSummarize generates text analysis
|
||||
Step 2 (every call): Settings.llm.acomplete formats analysis as JSON
|
||||
"""
|
||||
current_retries = await ctx.store.get("retries", default=0)
|
||||
await ctx.store.set("retries", current_retries + 1)
|
||||
|
||||
if current_retries >= self.max_retries:
|
||||
last_error = await ctx.store.get("last_error", default=None)
|
||||
logger.error(
|
||||
f"Max retries ({self.max_retries}) reached for {self.output_cls.__name__}"
|
||||
)
|
||||
return StopEvent(result={"error": last_error, "attempts": current_retries})
|
||||
|
||||
if isinstance(ev, StartEvent):
|
||||
# First call: run TreeSummarize to get analysis, store in context
|
||||
prompt = ev.get("prompt")
|
||||
texts = ev.get("texts")
|
||||
tone_name = ev.get("tone_name")
|
||||
if not prompt or not isinstance(texts, list):
|
||||
raise ValueError(
|
||||
"StartEvent must contain 'prompt' (str) and 'texts' (list)"
|
||||
)
|
||||
|
||||
summarizer = TreeSummarize(verbose=False)
|
||||
analysis = await summarizer.aget_response(
|
||||
prompt, texts, tone_name=tone_name
|
||||
)
|
||||
await ctx.store.set("analysis", str(analysis))
|
||||
reflection = ""
|
||||
else:
|
||||
# Retry: reuse analysis from context
|
||||
analysis = await ctx.store.get("analysis")
|
||||
if not analysis:
|
||||
raise RuntimeError("Internal error: analysis not found in context")
|
||||
|
||||
wrong_output = ev.wrong_output
|
||||
if len(wrong_output) > 2000:
|
||||
wrong_output = wrong_output[:2000] + "... [truncated]"
|
||||
reflection = (
|
||||
f"\n\nYour previous response could not be parsed:\n{wrong_output}\n\n"
|
||||
f"Error:\n{ev.error}\n\n"
|
||||
"Please try again. Return ONLY valid JSON matching the schema above, "
|
||||
"with no markdown formatting or extra text."
|
||||
)
|
||||
|
||||
# Step 2: Format analysis as JSON using LLM completion
|
||||
format_instructions = self.output_parser.format(
|
||||
"Please structure the above information in the following JSON format:"
|
||||
)
|
||||
|
||||
json_prompt = STRUCTURED_RESPONSE_PROMPT_TEMPLATE.format(
|
||||
analysis=analysis,
|
||||
format_instructions=format_instructions + reflection,
|
||||
)
|
||||
|
||||
# Network retries handled by OpenAILike (max_retries=3)
|
||||
response = await Settings.llm.acomplete(json_prompt)
|
||||
return ExtractionDone(output=response.text)
|
||||
|
||||
@step
|
||||
async def validate(
|
||||
self, ctx: Context, ev: ExtractionDone
|
||||
) -> StopEvent | ValidationErrorEvent:
|
||||
"""Validate extracted output against Pydantic schema."""
|
||||
raw_output = ev.output
|
||||
retries = await ctx.store.get("retries", default=0)
|
||||
|
||||
try:
|
||||
parsed = self.output_parser.parse(raw_output)
|
||||
if retries > 1:
|
||||
logger.info(
|
||||
f"LLM parse succeeded on attempt {retries}/{self.max_retries} "
|
||||
f"for {self.output_cls.__name__}"
|
||||
)
|
||||
return StopEvent(result={"success": parsed})
|
||||
|
||||
except (ValidationError, ValueError) as e:
|
||||
error_msg = self._format_error(e, raw_output)
|
||||
await ctx.store.set("last_error", error_msg)
|
||||
|
||||
logger.error(
|
||||
f"LLM parse error (attempt {retries}/{self.max_retries}): "
|
||||
f"{type(e).__name__}: {e}\nRaw response: {raw_output[:500]}"
|
||||
)
|
||||
|
||||
return ValidationErrorEvent(
|
||||
error=error_msg,
|
||||
wrong_output=raw_output,
|
||||
)
|
||||
|
||||
def _format_error(self, error: Exception, raw_output: str) -> str:
|
||||
"""Format error for LLM feedback."""
|
||||
if isinstance(error, ValidationError):
|
||||
error_messages = []
|
||||
for err in error.errors():
|
||||
field = ".".join(str(loc) for loc in err["loc"])
|
||||
error_messages.append(f"- {err['msg']} in field '{field}'")
|
||||
return "Schema validation errors:\n" + "\n".join(error_messages)
|
||||
else:
|
||||
return f"Parse error: {str(error)}"
|
||||
|
||||
|
||||
class LLM:
|
||||
def __init__(self, settings, temperature: float = 0.4, max_tokens: int = 2048):
|
||||
self.settings_obj = settings
|
||||
@@ -30,11 +200,12 @@ class LLM:
|
||||
self.temperature = temperature
|
||||
self.max_tokens = max_tokens
|
||||
|
||||
# Configure llamaindex Settings
|
||||
self._configure_llamaindex()
|
||||
|
||||
def _configure_llamaindex(self):
|
||||
"""Configure llamaindex Settings with OpenAILike LLM"""
|
||||
session_id = llm_session_id.get() or f"fallback-{uuid4().hex}"
|
||||
|
||||
Settings.llm = OpenAILike(
|
||||
model=self.model_name,
|
||||
api_base=self.url,
|
||||
@@ -44,6 +215,7 @@ class LLM:
|
||||
is_function_calling_model=False,
|
||||
temperature=self.temperature,
|
||||
max_tokens=self.max_tokens,
|
||||
additional_kwargs={"extra_body": {"litellm_session_id": session_id}},
|
||||
)
|
||||
|
||||
async def get_response(
|
||||
@@ -60,44 +232,38 @@ class LLM:
|
||||
texts: list[str],
|
||||
output_cls: Type[T],
|
||||
tone_name: str | None = None,
|
||||
timeout: int | None = None,
|
||||
) -> T:
|
||||
"""Get structured output from LLM for non-function-calling models"""
|
||||
logger = logging.getLogger(__name__)
|
||||
"""Get structured output from LLM with validation retry via Workflow."""
|
||||
if timeout is None:
|
||||
timeout = self.settings_obj.LLM_STRUCTURED_RESPONSE_TIMEOUT
|
||||
|
||||
summarizer = TreeSummarize(verbose=True)
|
||||
response = await summarizer.aget_response(prompt, texts, tone_name=tone_name)
|
||||
|
||||
output_parser = PydanticOutputParser(output_cls)
|
||||
|
||||
program = LLMTextCompletionProgram.from_defaults(
|
||||
output_parser=output_parser,
|
||||
prompt_template_str=STRUCTURED_RESPONSE_PROMPT_TEMPLATE,
|
||||
verbose=False,
|
||||
)
|
||||
|
||||
format_instructions = output_parser.format(
|
||||
"Please structure the above information in the following JSON format:"
|
||||
)
|
||||
|
||||
try:
|
||||
output = await program.acall(
|
||||
analysis=str(response), format_instructions=format_instructions
|
||||
async def run_workflow():
|
||||
workflow = StructuredOutputWorkflow(
|
||||
output_cls=output_cls,
|
||||
max_retries=self.settings_obj.LLM_PARSE_MAX_RETRIES + 1,
|
||||
timeout=timeout,
|
||||
)
|
||||
except ValidationError as e:
|
||||
# Extract the raw JSON from the error details
|
||||
errors = e.errors()
|
||||
if errors and "input" in errors[0]:
|
||||
raw_json = errors[0]["input"]
|
||||
logger.error(
|
||||
f"JSON validation failed for {output_cls.__name__}. "
|
||||
f"Full raw JSON output:\n{raw_json}\n"
|
||||
f"Validation errors: {errors}"
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
f"JSON validation failed for {output_cls.__name__}. "
|
||||
f"Validation errors: {errors}"
|
||||
)
|
||||
raise
|
||||
|
||||
return output
|
||||
result = await workflow.run(
|
||||
prompt=prompt,
|
||||
texts=texts,
|
||||
tone_name=tone_name,
|
||||
)
|
||||
|
||||
if "error" in result:
|
||||
error_msg = result["error"] or "Max retries exceeded"
|
||||
raise LLMParseError(
|
||||
output_cls=output_cls,
|
||||
error_msg=error_msg,
|
||||
attempts=result.get("attempts", 0),
|
||||
)
|
||||
|
||||
return result["success"]
|
||||
|
||||
return await retry(run_workflow)(
|
||||
retry_attempts=3,
|
||||
retry_backoff_interval=1.0,
|
||||
retry_backoff_max=30.0,
|
||||
retry_ignore_exc_types=(WorkflowTimeoutError,),
|
||||
)
|
||||
|
||||
@@ -309,6 +309,7 @@ class PipelineMainFile(PipelineMainBase):
|
||||
transcript,
|
||||
on_long_summary_callback=self.on_long_summary,
|
||||
on_short_summary_callback=self.on_short_summary,
|
||||
on_action_items_callback=self.on_action_items,
|
||||
empty_pipeline=self.empty_pipeline,
|
||||
logger=self.logger,
|
||||
)
|
||||
@@ -340,7 +341,6 @@ async def task_send_webhook_if_needed(*, transcript_id: str):
|
||||
@asynctask
|
||||
async def task_pipeline_file_process(*, transcript_id: str):
|
||||
"""Celery task for file pipeline processing"""
|
||||
|
||||
transcript = await transcripts_controller.get_by_id(transcript_id)
|
||||
if not transcript:
|
||||
raise Exception(f"Transcript {transcript_id} not found")
|
||||
|
||||
@@ -27,6 +27,7 @@ from reflector.db.recordings import recordings_controller
|
||||
from reflector.db.rooms import rooms_controller
|
||||
from reflector.db.transcripts import (
|
||||
Transcript,
|
||||
TranscriptActionItems,
|
||||
TranscriptDuration,
|
||||
TranscriptFinalLongSummary,
|
||||
TranscriptFinalShortSummary,
|
||||
@@ -306,6 +307,23 @@ class PipelineMainBase(PipelineRunner[PipelineMessage], Generic[PipelineMessage]
|
||||
data=final_short_summary,
|
||||
)
|
||||
|
||||
@broadcast_to_sockets
|
||||
async def on_action_items(self, data):
|
||||
action_items = TranscriptActionItems(action_items=data.action_items)
|
||||
async with self.transaction():
|
||||
transcript = await self.get_transcript()
|
||||
await transcripts_controller.update(
|
||||
transcript,
|
||||
{
|
||||
"action_items": action_items.action_items,
|
||||
},
|
||||
)
|
||||
return await transcripts_controller.append_event(
|
||||
transcript=transcript,
|
||||
event="ACTION_ITEMS",
|
||||
data=action_items,
|
||||
)
|
||||
|
||||
@broadcast_to_sockets
|
||||
async def on_duration(self, data):
|
||||
async with self.transaction():
|
||||
@@ -465,6 +483,7 @@ class PipelineMainFinalSummaries(PipelineMainFromTopics):
|
||||
transcript=self._transcript,
|
||||
callback=self.on_long_summary,
|
||||
on_short_summary=self.on_short_summary,
|
||||
on_action_items=self.on_action_items,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@@ -422,7 +422,15 @@ class PipelineMainMultitrack(PipelineMainBase):
|
||||
# Open all containers with cleanup guaranteed
|
||||
for i, url in enumerate(valid_track_urls):
|
||||
try:
|
||||
c = av.open(url)
|
||||
c = av.open(
|
||||
url,
|
||||
options={
|
||||
# it's trying to stream from s3 by default
|
||||
"reconnect": "1",
|
||||
"reconnect_streamed": "1",
|
||||
"reconnect_delay_max": "5",
|
||||
},
|
||||
)
|
||||
containers.append(c)
|
||||
except Exception as e:
|
||||
self.logger.warning(
|
||||
@@ -451,6 +459,8 @@ class PipelineMainMultitrack(PipelineMainBase):
|
||||
frame = next(dec)
|
||||
except StopIteration:
|
||||
active[i] = False
|
||||
# causes stream to move on / unclogs memory
|
||||
inputs[i].push(None)
|
||||
continue
|
||||
|
||||
if frame.sample_rate != target_sample_rate:
|
||||
@@ -470,8 +480,6 @@ class PipelineMainMultitrack(PipelineMainBase):
|
||||
mixed.time_base = Fraction(1, target_sample_rate)
|
||||
await writer.push(mixed)
|
||||
|
||||
for in_ctx in inputs:
|
||||
in_ctx.push(None)
|
||||
while True:
|
||||
try:
|
||||
mixed = sink.pull()
|
||||
@@ -764,6 +772,7 @@ class PipelineMainMultitrack(PipelineMainBase):
|
||||
transcript,
|
||||
on_long_summary_callback=self.on_long_summary,
|
||||
on_short_summary_callback=self.on_short_summary,
|
||||
on_action_items_callback=self.on_action_items,
|
||||
empty_pipeline=self.empty_pipeline,
|
||||
logger=self.logger,
|
||||
)
|
||||
|
||||
@@ -89,6 +89,7 @@ async def generate_summaries(
|
||||
*,
|
||||
on_long_summary_callback: Callable,
|
||||
on_short_summary_callback: Callable,
|
||||
on_action_items_callback: Callable,
|
||||
empty_pipeline: EmptyPipeline,
|
||||
logger: structlog.BoundLogger,
|
||||
):
|
||||
@@ -96,11 +97,14 @@ async def generate_summaries(
|
||||
logger.warning("No topics for summary generation")
|
||||
return
|
||||
|
||||
processor = TranscriptFinalSummaryProcessor(
|
||||
transcript=transcript,
|
||||
callback=on_long_summary_callback,
|
||||
on_short_summary=on_short_summary_callback,
|
||||
)
|
||||
processor_kwargs = {
|
||||
"transcript": transcript,
|
||||
"callback": on_long_summary_callback,
|
||||
"on_short_summary": on_short_summary_callback,
|
||||
"on_action_items": on_action_items_callback,
|
||||
}
|
||||
|
||||
processor = TranscriptFinalSummaryProcessor(**processor_kwargs)
|
||||
processor.set_pipeline(empty_pipeline)
|
||||
|
||||
for topic in topics:
|
||||
|
||||
@@ -96,6 +96,36 @@ RECAP_PROMPT = dedent(
|
||||
"""
|
||||
).strip()
|
||||
|
||||
ACTION_ITEMS_PROMPT = dedent(
|
||||
"""
|
||||
Identify action items from this meeting transcript. Your goal is to identify what was decided and what needs to happen next.
|
||||
|
||||
Look for:
|
||||
|
||||
1. **Decisions Made**: Any decisions, choices, or conclusions reached during the meeting. For each decision:
|
||||
- What was decided? (be specific)
|
||||
- Who made the decision or was involved? (use actual participant names)
|
||||
- Why was this decision made? (key factors, reasoning, or rationale)
|
||||
|
||||
2. **Next Steps / Action Items**: Any tasks, follow-ups, or actions that were mentioned or assigned. For each action item:
|
||||
- What specific task needs to be done? (be concrete and actionable)
|
||||
- Who is responsible? (use actual participant names if mentioned, or "team" if unclear)
|
||||
- When is it due? (any deadlines, timeframes, or "by next meeting" type commitments)
|
||||
- What context is needed? (any additional details that help understand the task)
|
||||
|
||||
Guidelines:
|
||||
- Be thorough and identify all action items, even if they seem minor
|
||||
- Include items that were agreed upon, assigned, or committed to
|
||||
- Include decisions even if they seem obvious or implicit
|
||||
- If someone says "I'll do X" or "We should do Y", that's an action item
|
||||
- If someone says "Let's go with option A", that's a decision
|
||||
- Use the exact participant names from the transcript
|
||||
- If no participant name is mentioned, you can leave assigned_to/decided_by as null
|
||||
|
||||
Only return empty lists if the transcript contains NO decisions and NO action items whatsoever.
|
||||
"""
|
||||
).strip()
|
||||
|
||||
STRUCTURED_RESPONSE_PROMPT_TEMPLATE = dedent(
|
||||
"""
|
||||
Based on the following analysis, provide the information in the requested JSON format:
|
||||
@@ -155,6 +185,53 @@ class SubjectsResponse(BaseModel):
|
||||
)
|
||||
|
||||
|
||||
class ActionItem(BaseModel):
|
||||
"""A single action item from the meeting"""
|
||||
|
||||
task: str = Field(description="The task or action item to be completed")
|
||||
assigned_to: str | None = Field(
|
||||
default=None, description="Person or team assigned to this task (name)"
|
||||
)
|
||||
assigned_to_participant_id: str | None = Field(
|
||||
default=None, description="Participant ID if assigned_to matches a participant"
|
||||
)
|
||||
deadline: str | None = Field(
|
||||
default=None, description="Deadline or timeframe mentioned for this task"
|
||||
)
|
||||
context: str | None = Field(
|
||||
default=None, description="Additional context or notes about this task"
|
||||
)
|
||||
|
||||
|
||||
class Decision(BaseModel):
|
||||
"""A decision made during the meeting"""
|
||||
|
||||
decision: str = Field(description="What was decided")
|
||||
rationale: str | None = Field(
|
||||
default=None,
|
||||
description="Reasoning or key factors that influenced this decision",
|
||||
)
|
||||
decided_by: str | None = Field(
|
||||
default=None, description="Person or group who made the decision (name)"
|
||||
)
|
||||
decided_by_participant_id: str | None = Field(
|
||||
default=None, description="Participant ID if decided_by matches a participant"
|
||||
)
|
||||
|
||||
|
||||
class ActionItemsResponse(BaseModel):
|
||||
"""Pydantic model for identified action items"""
|
||||
|
||||
decisions: list[Decision] = Field(
|
||||
default_factory=list,
|
||||
description="List of decisions made during the meeting",
|
||||
)
|
||||
next_steps: list[ActionItem] = Field(
|
||||
default_factory=list,
|
||||
description="List of action items and next steps to be taken",
|
||||
)
|
||||
|
||||
|
||||
class SummaryBuilder:
|
||||
def __init__(self, llm: LLM, filename: str | None = None, logger=None) -> None:
|
||||
self.transcript: str | None = None
|
||||
@@ -166,6 +243,8 @@ class SummaryBuilder:
|
||||
self.model_name: str = llm.model_name
|
||||
self.logger = logger or structlog.get_logger()
|
||||
self.participant_instructions: str | None = None
|
||||
self.action_items: ActionItemsResponse | None = None
|
||||
self.participant_name_to_id: dict[str, str] = {}
|
||||
if filename:
|
||||
self.read_transcript_from_file(filename)
|
||||
|
||||
@@ -189,13 +268,20 @@ class SummaryBuilder:
|
||||
self.llm = llm
|
||||
|
||||
async def _get_structured_response(
|
||||
self, prompt: str, output_cls: Type[T], tone_name: str | None = None
|
||||
self,
|
||||
prompt: str,
|
||||
output_cls: Type[T],
|
||||
tone_name: str | None = None,
|
||||
timeout: int | None = None,
|
||||
) -> T:
|
||||
"""Generic function to get structured output from LLM for non-function-calling models."""
|
||||
# Add participant instructions to the prompt if available
|
||||
enhanced_prompt = self._enhance_prompt_with_participants(prompt)
|
||||
return await self.llm.get_structured_response(
|
||||
enhanced_prompt, [self.transcript], output_cls, tone_name=tone_name
|
||||
enhanced_prompt,
|
||||
[self.transcript],
|
||||
output_cls,
|
||||
tone_name=tone_name,
|
||||
timeout=timeout,
|
||||
)
|
||||
|
||||
async def _get_response(
|
||||
@@ -216,11 +302,19 @@ class SummaryBuilder:
|
||||
# Participants
|
||||
# ----------------------------------------------------------------------------
|
||||
|
||||
def set_known_participants(self, participants: list[str]) -> None:
|
||||
def set_known_participants(
|
||||
self,
|
||||
participants: list[str],
|
||||
participant_name_to_id: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
"""
|
||||
Set known participants directly without LLM identification.
|
||||
This is used when participants are already identified and stored.
|
||||
They are appended at the end of the transcript, providing more context for the assistant.
|
||||
|
||||
Args:
|
||||
participants: List of participant names
|
||||
participant_name_to_id: Optional mapping of participant names to their IDs
|
||||
"""
|
||||
if not participants:
|
||||
self.logger.warning("No participants provided")
|
||||
@@ -231,10 +325,12 @@ class SummaryBuilder:
|
||||
participants=participants,
|
||||
)
|
||||
|
||||
if participant_name_to_id:
|
||||
self.participant_name_to_id = participant_name_to_id
|
||||
|
||||
participants_md = self.format_list_md(participants)
|
||||
self.transcript += f"\n\n# Participants\n\n{participants_md}"
|
||||
|
||||
# Set instructions that will be automatically added to all prompts
|
||||
participants_list = ", ".join(participants)
|
||||
self.participant_instructions = dedent(
|
||||
f"""
|
||||
@@ -413,6 +509,92 @@ class SummaryBuilder:
|
||||
self.recap = str(recap_response)
|
||||
self.logger.info(f"Quick recap: {self.recap}")
|
||||
|
||||
def _map_participant_names_to_ids(
|
||||
self, response: ActionItemsResponse
|
||||
) -> ActionItemsResponse:
|
||||
"""Map participant names in action items to participant IDs."""
|
||||
if not self.participant_name_to_id:
|
||||
return response
|
||||
|
||||
decisions = []
|
||||
for decision in response.decisions:
|
||||
new_decision = decision.model_copy()
|
||||
if (
|
||||
decision.decided_by
|
||||
and decision.decided_by in self.participant_name_to_id
|
||||
):
|
||||
new_decision.decided_by_participant_id = self.participant_name_to_id[
|
||||
decision.decided_by
|
||||
]
|
||||
decisions.append(new_decision)
|
||||
|
||||
next_steps = []
|
||||
for item in response.next_steps:
|
||||
new_item = item.model_copy()
|
||||
if item.assigned_to and item.assigned_to in self.participant_name_to_id:
|
||||
new_item.assigned_to_participant_id = self.participant_name_to_id[
|
||||
item.assigned_to
|
||||
]
|
||||
next_steps.append(new_item)
|
||||
|
||||
return ActionItemsResponse(decisions=decisions, next_steps=next_steps)
|
||||
|
||||
async def identify_action_items(self) -> ActionItemsResponse | None:
|
||||
"""Identify action items (decisions and next steps) from the transcript."""
|
||||
self.logger.info("--- identify action items using TreeSummarize")
|
||||
|
||||
if not self.transcript:
|
||||
self.logger.warning(
|
||||
"No transcript available for action items identification"
|
||||
)
|
||||
self.action_items = None
|
||||
return None
|
||||
|
||||
action_items_prompt = ACTION_ITEMS_PROMPT
|
||||
|
||||
try:
|
||||
response = await self._get_structured_response(
|
||||
action_items_prompt,
|
||||
ActionItemsResponse,
|
||||
tone_name="Action item identifier",
|
||||
timeout=settings.LLM_STRUCTURED_RESPONSE_TIMEOUT,
|
||||
)
|
||||
|
||||
response = self._map_participant_names_to_ids(response)
|
||||
|
||||
self.action_items = response
|
||||
self.logger.info(
|
||||
f"Identified {len(response.decisions)} decisions and {len(response.next_steps)} action items",
|
||||
decisions_count=len(response.decisions),
|
||||
next_steps_count=len(response.next_steps),
|
||||
)
|
||||
|
||||
if response.decisions:
|
||||
self.logger.debug(
|
||||
"Decisions identified",
|
||||
decisions=[d.decision for d in response.decisions],
|
||||
)
|
||||
if response.next_steps:
|
||||
self.logger.debug(
|
||||
"Action items identified",
|
||||
tasks=[item.task for item in response.next_steps],
|
||||
)
|
||||
if not response.decisions and not response.next_steps:
|
||||
self.logger.warning(
|
||||
"No action items identified from transcript",
|
||||
transcript_length=len(self.transcript),
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(
|
||||
f"Error identifying action items: {e}",
|
||||
exc_info=True,
|
||||
)
|
||||
self.action_items = None
|
||||
return None
|
||||
|
||||
async def generate_summary(self, only_subjects: bool = False) -> None:
|
||||
"""
|
||||
Generate summary by extracting subjects, creating summaries for each, and generating a recap.
|
||||
@@ -424,6 +606,7 @@ class SummaryBuilder:
|
||||
|
||||
await self.generate_subject_summaries()
|
||||
await self.generate_recap()
|
||||
await self.identify_action_items()
|
||||
|
||||
# ----------------------------------------------------------------------------
|
||||
# Markdown
|
||||
@@ -526,8 +709,6 @@ if __name__ == "__main__":
|
||||
if args.summary:
|
||||
await sm.generate_summary()
|
||||
|
||||
# Note: action items generation has been removed
|
||||
|
||||
print("")
|
||||
print("-" * 80)
|
||||
print("")
|
||||
|
||||
@@ -1,7 +1,12 @@
|
||||
from reflector.llm import LLM
|
||||
from reflector.processors.base import Processor
|
||||
from reflector.processors.summary.summary_builder import SummaryBuilder
|
||||
from reflector.processors.types import FinalLongSummary, FinalShortSummary, TitleSummary
|
||||
from reflector.processors.types import (
|
||||
ActionItems,
|
||||
FinalLongSummary,
|
||||
FinalShortSummary,
|
||||
TitleSummary,
|
||||
)
|
||||
from reflector.settings import settings
|
||||
|
||||
|
||||
@@ -27,15 +32,20 @@ class TranscriptFinalSummaryProcessor(Processor):
|
||||
builder = SummaryBuilder(self.llm, logger=self.logger)
|
||||
builder.set_transcript(text)
|
||||
|
||||
# Use known participants if available, otherwise identify them
|
||||
if self.transcript and self.transcript.participants:
|
||||
# Extract participant names from the stored participants
|
||||
participant_names = [p.name for p in self.transcript.participants if p.name]
|
||||
if participant_names:
|
||||
self.logger.info(
|
||||
f"Using {len(participant_names)} known participants from transcript"
|
||||
)
|
||||
builder.set_known_participants(participant_names)
|
||||
participant_name_to_id = {
|
||||
p.name: p.id
|
||||
for p in self.transcript.participants
|
||||
if p.name and p.id
|
||||
}
|
||||
builder.set_known_participants(
|
||||
participant_names, participant_name_to_id=participant_name_to_id
|
||||
)
|
||||
else:
|
||||
self.logger.info(
|
||||
"Participants field exists but is empty, identifying participants"
|
||||
@@ -63,7 +73,6 @@ class TranscriptFinalSummaryProcessor(Processor):
|
||||
self.logger.warning("No summary to output")
|
||||
return
|
||||
|
||||
# build the speakermap from the transcript
|
||||
speakermap = {}
|
||||
if self.transcript:
|
||||
speakermap = {
|
||||
@@ -76,8 +85,6 @@ class TranscriptFinalSummaryProcessor(Processor):
|
||||
speakermap=speakermap,
|
||||
)
|
||||
|
||||
# build the transcript as a single string
|
||||
# Replace speaker IDs with actual participant names if available
|
||||
text_transcript = []
|
||||
unique_speakers = set()
|
||||
for topic in self.chunks:
|
||||
@@ -111,4 +118,9 @@ class TranscriptFinalSummaryProcessor(Processor):
|
||||
)
|
||||
await self.emit(final_short_summary, name="short_summary")
|
||||
|
||||
if self.builder and self.builder.action_items:
|
||||
action_items = self.builder.action_items.model_dump()
|
||||
action_items = ActionItems(action_items=action_items)
|
||||
await self.emit(action_items, name="action_items")
|
||||
|
||||
await self.emit(final_long_summary)
|
||||
|
||||
@@ -78,7 +78,11 @@ class TranscriptTopicDetectorProcessor(Processor):
|
||||
"""
|
||||
prompt = TOPIC_PROMPT.format(text=text)
|
||||
response = await self.llm.get_structured_response(
|
||||
prompt, [text], TopicResponse, tone_name="Topic analyzer"
|
||||
prompt,
|
||||
[text],
|
||||
TopicResponse,
|
||||
tone_name="Topic analyzer",
|
||||
timeout=settings.LLM_STRUCTURED_RESPONSE_TIMEOUT,
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
@@ -264,6 +264,10 @@ class FinalShortSummary(BaseModel):
|
||||
duration: float
|
||||
|
||||
|
||||
class ActionItems(BaseModel):
|
||||
action_items: dict # JSON-serializable dict from ActionItemsResponse
|
||||
|
||||
|
||||
class FinalTitle(BaseModel):
|
||||
title: str
|
||||
|
||||
|
||||
@@ -160,7 +160,10 @@ def dispatch_transcript_processing(config: ProcessingConfig) -> AsyncResult:
|
||||
def task_is_scheduled_or_active(task_name: str, **kwargs):
|
||||
inspect = celery.current_app.control.inspect()
|
||||
|
||||
for worker, tasks in (inspect.scheduled() | inspect.active()).items():
|
||||
scheduled = inspect.scheduled() or {}
|
||||
active = inspect.active() or {}
|
||||
all = scheduled | active
|
||||
for worker, tasks in all.items():
|
||||
for task in tasks:
|
||||
if task["name"] == task_name and task["kwargs"] == kwargs:
|
||||
return True
|
||||
|
||||
@@ -74,6 +74,13 @@ class Settings(BaseSettings):
|
||||
LLM_API_KEY: str | None = None
|
||||
LLM_CONTEXT_WINDOW: int = 16000
|
||||
|
||||
LLM_PARSE_MAX_RETRIES: int = (
|
||||
3 # Max retries for JSON/validation errors (total attempts = retries + 1)
|
||||
)
|
||||
LLM_STRUCTURED_RESPONSE_TIMEOUT: int = (
|
||||
300 # Timeout in seconds for structured responses (5 minutes)
|
||||
)
|
||||
|
||||
# Diarization
|
||||
DIARIZATION_ENABLED: bool = True
|
||||
DIARIZATION_BACKEND: str = "modal"
|
||||
|
||||
@@ -31,6 +31,7 @@ class DailyClient(VideoPlatformClient):
|
||||
PLATFORM_NAME: Platform = "daily"
|
||||
TIMESTAMP_FORMAT = "%Y%m%d%H%M%S"
|
||||
RECORDING_NONE: RecordingType = "none"
|
||||
RECORDING_LOCAL: RecordingType = "local"
|
||||
RECORDING_CLOUD: RecordingType = "cloud"
|
||||
|
||||
def __init__(self, config: VideoPlatformConfig):
|
||||
@@ -54,19 +55,23 @@ class DailyClient(VideoPlatformClient):
|
||||
timestamp = datetime.now().strftime(self.TIMESTAMP_FORMAT)
|
||||
room_name = f"{room_name_prefix}{ROOM_PREFIX_SEPARATOR}{timestamp}"
|
||||
|
||||
enable_recording = None
|
||||
if room.recording_type == self.RECORDING_LOCAL:
|
||||
enable_recording = "local"
|
||||
elif room.recording_type == self.RECORDING_CLOUD:
|
||||
enable_recording = "raw-tracks"
|
||||
|
||||
properties = RoomProperties(
|
||||
enable_recording="raw-tracks"
|
||||
if room.recording_type != self.RECORDING_NONE
|
||||
else False,
|
||||
enable_recording=enable_recording,
|
||||
enable_chat=True,
|
||||
enable_screenshare=True,
|
||||
enable_knocking=room.is_locked,
|
||||
start_video_off=False,
|
||||
start_audio_off=False,
|
||||
exp=int(end_date.timestamp()),
|
||||
)
|
||||
|
||||
# Only configure recordings_bucket if recording is enabled
|
||||
if room.recording_type != self.RECORDING_NONE:
|
||||
if room.recording_type == self.RECORDING_CLOUD:
|
||||
daily_storage = get_dailyco_storage()
|
||||
assert daily_storage.bucket_name, "S3 bucket must be configured"
|
||||
properties.recordings_bucket = RecordingsBucketConfig(
|
||||
@@ -172,15 +177,16 @@ class DailyClient(VideoPlatformClient):
|
||||
async def create_meeting_token(
|
||||
self,
|
||||
room_name: DailyRoomName,
|
||||
enable_recording: bool,
|
||||
start_cloud_recording: bool,
|
||||
enable_recording_ui: bool,
|
||||
user_id: NonEmptyString | None = None,
|
||||
is_owner: bool = False,
|
||||
) -> NonEmptyString:
|
||||
properties = MeetingTokenProperties(
|
||||
room_name=room_name,
|
||||
user_id=user_id,
|
||||
start_cloud_recording=enable_recording,
|
||||
enable_recording_ui=False,
|
||||
start_cloud_recording=start_cloud_recording,
|
||||
enable_recording_ui=enable_recording_ui,
|
||||
is_owner=is_owner,
|
||||
)
|
||||
request = CreateMeetingTokenRequest(properties=properties)
|
||||
|
||||
@@ -89,7 +89,7 @@ class CreateRoom(BaseModel):
|
||||
ics_url: Optional[str] = None
|
||||
ics_fetch_interval: int = 300
|
||||
ics_enabled: bool = False
|
||||
platform: Optional[Platform] = None
|
||||
platform: Platform
|
||||
|
||||
|
||||
class UpdateRoom(BaseModel):
|
||||
@@ -248,7 +248,7 @@ async def rooms_create(
|
||||
ics_url=room.ics_url,
|
||||
ics_fetch_interval=room.ics_fetch_interval,
|
||||
ics_enabled=room.ics_enabled,
|
||||
platform=room.platform or settings.DEFAULT_VIDEO_PLATFORM,
|
||||
platform=room.platform,
|
||||
)
|
||||
|
||||
|
||||
@@ -310,6 +310,22 @@ async def rooms_create_meeting(
|
||||
room=room, current_time=current_time
|
||||
)
|
||||
|
||||
if meeting is not None:
|
||||
settings_match = (
|
||||
meeting.is_locked == room.is_locked
|
||||
and meeting.room_mode == room.room_mode
|
||||
and meeting.recording_type == room.recording_type
|
||||
and meeting.recording_trigger == room.recording_trigger
|
||||
and meeting.platform == room.platform
|
||||
)
|
||||
if not settings_match:
|
||||
logger.info(
|
||||
f"Room settings changed for {room_name}, creating new meeting",
|
||||
room_id=room.id,
|
||||
old_meeting_id=meeting.id,
|
||||
)
|
||||
meeting = None
|
||||
|
||||
if meeting is None:
|
||||
end_date = current_time + timedelta(hours=8)
|
||||
|
||||
@@ -549,21 +565,16 @@ async def rooms_join_meeting(
|
||||
if meeting.end_date <= current_time:
|
||||
raise HTTPException(status_code=400, detail="Meeting has ended")
|
||||
|
||||
if meeting.platform == "daily":
|
||||
if meeting.platform == "daily" and user_id is not None:
|
||||
client = create_platform_client(meeting.platform)
|
||||
enable_recording = room.recording_trigger != "none"
|
||||
token = await client.create_meeting_token(
|
||||
meeting.room_name,
|
||||
enable_recording=enable_recording,
|
||||
start_cloud_recording=meeting.recording_type == "cloud",
|
||||
enable_recording_ui=meeting.recording_type == "local",
|
||||
user_id=user_id,
|
||||
is_owner=user_id == room.user_id,
|
||||
)
|
||||
meeting = meeting.model_copy()
|
||||
meeting.room_url = add_query_param(meeting.room_url, "t", token)
|
||||
if meeting.host_room_url:
|
||||
meeting.host_room_url = add_query_param(meeting.host_room_url, "t", token)
|
||||
|
||||
if user_id != room.user_id and meeting.platform == "whereby":
|
||||
meeting.host_room_url = ""
|
||||
|
||||
return meeting
|
||||
|
||||
@@ -17,6 +17,7 @@ from pydantic import (
|
||||
import reflector.auth as auth
|
||||
from reflector.db import get_database
|
||||
from reflector.db.recordings import recordings_controller
|
||||
from reflector.db.rooms import rooms_controller
|
||||
from reflector.db.search import (
|
||||
DEFAULT_SEARCH_LIMIT,
|
||||
SearchLimit,
|
||||
@@ -37,6 +38,7 @@ from reflector.db.transcripts import (
|
||||
TranscriptTopic,
|
||||
transcripts_controller,
|
||||
)
|
||||
from reflector.db.users import user_controller
|
||||
from reflector.processors.types import Transcript as ProcessorTranscript
|
||||
from reflector.processors.types import Word
|
||||
from reflector.schemas.transcript_formats import TranscriptFormat, TranscriptSegment
|
||||
@@ -111,8 +113,12 @@ class GetTranscriptMinimal(BaseModel):
|
||||
audio_deleted: bool | None = None
|
||||
|
||||
|
||||
class TranscriptParticipantWithEmail(TranscriptParticipant):
|
||||
email: str | None = None
|
||||
|
||||
|
||||
class GetTranscriptWithParticipants(GetTranscriptMinimal):
|
||||
participants: list[TranscriptParticipant] | None
|
||||
participants: list[TranscriptParticipantWithEmail] | None
|
||||
|
||||
|
||||
class GetTranscriptWithText(GetTranscriptWithParticipants):
|
||||
@@ -468,6 +474,23 @@ async def transcript_get(
|
||||
|
||||
is_multitrack = await _get_is_multitrack(transcript)
|
||||
|
||||
room_name = None
|
||||
if transcript.room_id:
|
||||
room = await rooms_controller.get_by_id(transcript.room_id)
|
||||
room_name = room.name if room else None
|
||||
|
||||
participants = []
|
||||
if transcript.participants:
|
||||
user_ids = [p.user_id for p in transcript.participants if p.user_id is not None]
|
||||
users_dict = await user_controller.get_by_ids(user_ids) if user_ids else {}
|
||||
for p in transcript.participants:
|
||||
user = users_dict.get(p.user_id) if p.user_id else None
|
||||
participants.append(
|
||||
TranscriptParticipantWithEmail(
|
||||
**p.model_dump(), email=user.email if user else None
|
||||
)
|
||||
)
|
||||
|
||||
base_data = {
|
||||
"id": transcript.id,
|
||||
"user_id": transcript.user_id,
|
||||
@@ -478,6 +501,7 @@ async def transcript_get(
|
||||
"title": transcript.title,
|
||||
"short_summary": transcript.short_summary,
|
||||
"long_summary": transcript.long_summary,
|
||||
"action_items": transcript.action_items,
|
||||
"created_at": transcript.created_at,
|
||||
"share_mode": transcript.share_mode,
|
||||
"source_language": transcript.source_language,
|
||||
@@ -486,8 +510,9 @@ async def transcript_get(
|
||||
"meeting_id": transcript.meeting_id,
|
||||
"source_kind": transcript.source_kind,
|
||||
"room_id": transcript.room_id,
|
||||
"room_name": room_name,
|
||||
"audio_deleted": transcript.audio_deleted,
|
||||
"participants": transcript.participants,
|
||||
"participants": participants,
|
||||
}
|
||||
|
||||
if transcript_format == "text":
|
||||
|
||||
@@ -38,6 +38,10 @@ else:
|
||||
"task": "reflector.worker.process.reprocess_failed_recordings",
|
||||
"schedule": crontab(hour=5, minute=0), # Midnight EST
|
||||
},
|
||||
"reprocess_failed_daily_recordings": {
|
||||
"task": "reflector.worker.process.reprocess_failed_daily_recordings",
|
||||
"schedule": crontab(hour=5, minute=0), # Midnight EST
|
||||
},
|
||||
"poll_daily_recordings": {
|
||||
"task": "reflector.worker.process.poll_daily_recordings",
|
||||
"schedule": 180.0, # Every 3 minutes (configurable lookback window)
|
||||
|
||||
@@ -12,7 +12,7 @@ from celery import shared_task
|
||||
from celery.utils.log import get_task_logger
|
||||
from pydantic import ValidationError
|
||||
|
||||
from reflector.dailyco_api import RecordingResponse
|
||||
from reflector.dailyco_api import FinishedRecordingResponse, RecordingResponse
|
||||
from reflector.db.daily_participant_sessions import (
|
||||
DailyParticipantSession,
|
||||
daily_participant_sessions_controller,
|
||||
@@ -322,16 +322,38 @@ async def poll_daily_recordings():
|
||||
)
|
||||
return
|
||||
|
||||
recording_ids = [rec.id for rec in api_recordings]
|
||||
finished_recordings: List[FinishedRecordingResponse] = []
|
||||
for rec in api_recordings:
|
||||
finished = rec.to_finished()
|
||||
if finished is None:
|
||||
logger.debug(
|
||||
"Skipping unfinished recording",
|
||||
recording_id=rec.id,
|
||||
room_name=rec.room_name,
|
||||
status=rec.status,
|
||||
)
|
||||
continue
|
||||
finished_recordings.append(finished)
|
||||
|
||||
if not finished_recordings:
|
||||
logger.debug(
|
||||
"No finished recordings found from Daily.co API",
|
||||
total_api_count=len(api_recordings),
|
||||
)
|
||||
return
|
||||
|
||||
recording_ids = [rec.id for rec in finished_recordings]
|
||||
existing_recordings = await recordings_controller.get_by_ids(recording_ids)
|
||||
existing_ids = {rec.id for rec in existing_recordings}
|
||||
|
||||
missing_recordings = [rec for rec in api_recordings if rec.id not in existing_ids]
|
||||
missing_recordings = [
|
||||
rec for rec in finished_recordings if rec.id not in existing_ids
|
||||
]
|
||||
|
||||
if not missing_recordings:
|
||||
logger.debug(
|
||||
"All recordings already in DB",
|
||||
api_count=len(api_recordings),
|
||||
api_count=len(finished_recordings),
|
||||
existing_count=len(existing_recordings),
|
||||
)
|
||||
return
|
||||
@@ -339,7 +361,7 @@ async def poll_daily_recordings():
|
||||
logger.info(
|
||||
"Found recordings missing from DB",
|
||||
missing_count=len(missing_recordings),
|
||||
total_api_count=len(api_recordings),
|
||||
total_api_count=len(finished_recordings),
|
||||
existing_count=len(existing_recordings),
|
||||
)
|
||||
|
||||
@@ -649,7 +671,7 @@ async def reprocess_failed_recordings():
|
||||
Find recordings in Whereby S3 bucket and check if they have proper transcriptions.
|
||||
If not, requeue them for processing.
|
||||
|
||||
Note: Daily.co recordings are processed via webhooks, not this cron job.
|
||||
Note: Daily.co multitrack recordings are handled by reprocess_failed_daily_recordings.
|
||||
"""
|
||||
logger.info("Checking Whereby recordings that need processing or reprocessing")
|
||||
|
||||
@@ -702,6 +724,103 @@ async def reprocess_failed_recordings():
|
||||
return reprocessed_count
|
||||
|
||||
|
||||
@shared_task
|
||||
@asynctask
|
||||
async def reprocess_failed_daily_recordings():
|
||||
"""
|
||||
Find Daily.co multitrack recordings in the database and check if they have proper transcriptions.
|
||||
If not, requeue them for processing.
|
||||
"""
|
||||
logger.info(
|
||||
"Checking Daily.co multitrack recordings that need processing or reprocessing"
|
||||
)
|
||||
|
||||
if not settings.DAILYCO_STORAGE_AWS_BUCKET_NAME:
|
||||
logger.debug(
|
||||
"DAILYCO_STORAGE_AWS_BUCKET_NAME not configured; skipping Daily recording reprocessing"
|
||||
)
|
||||
return 0
|
||||
|
||||
bucket_name = settings.DAILYCO_STORAGE_AWS_BUCKET_NAME
|
||||
reprocessed_count = 0
|
||||
|
||||
try:
|
||||
multitrack_recordings = (
|
||||
await recordings_controller.get_multitrack_needing_reprocessing(bucket_name)
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Found multitrack recordings needing reprocessing",
|
||||
count=len(multitrack_recordings),
|
||||
bucket=bucket_name,
|
||||
)
|
||||
|
||||
for recording in multitrack_recordings:
|
||||
if not recording.meeting_id:
|
||||
logger.debug(
|
||||
"Skipping recording without meeting_id",
|
||||
recording_id=recording.id,
|
||||
)
|
||||
continue
|
||||
|
||||
meeting = await meetings_controller.get_by_id(recording.meeting_id)
|
||||
if not meeting:
|
||||
logger.warning(
|
||||
"Meeting not found for recording",
|
||||
recording_id=recording.id,
|
||||
meeting_id=recording.meeting_id,
|
||||
)
|
||||
continue
|
||||
|
||||
transcript = None
|
||||
try:
|
||||
transcript = await transcripts_controller.get_by_recording_id(
|
||||
recording.id
|
||||
)
|
||||
except ValidationError:
|
||||
await transcripts_controller.remove_by_recording_id(recording.id)
|
||||
logger.warning(
|
||||
"Removed invalid transcript for recording",
|
||||
recording_id=recording.id,
|
||||
)
|
||||
|
||||
if not recording.track_keys:
|
||||
logger.warning(
|
||||
"Recording has no track_keys, cannot reprocess",
|
||||
recording_id=recording.id,
|
||||
)
|
||||
continue
|
||||
|
||||
logger.info(
|
||||
"Queueing Daily recording for reprocessing",
|
||||
recording_id=recording.id,
|
||||
room_name=meeting.room_name,
|
||||
track_count=len(recording.track_keys),
|
||||
transcript_status=transcript.status if transcript else None,
|
||||
)
|
||||
|
||||
process_multitrack_recording.delay(
|
||||
bucket_name=bucket_name,
|
||||
daily_room_name=meeting.room_name,
|
||||
recording_id=recording.id,
|
||||
track_keys=recording.track_keys,
|
||||
)
|
||||
reprocessed_count += 1
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Error checking Daily multitrack recordings",
|
||||
error=str(e),
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Daily reprocessing complete",
|
||||
requeued_count=reprocessed_count,
|
||||
)
|
||||
return reprocessed_count
|
||||
|
||||
|
||||
@shared_task
|
||||
@asynctask
|
||||
async def trigger_daily_reconciliation() -> None:
|
||||
|
||||
@@ -123,6 +123,7 @@ async def send_transcript_webhook(
|
||||
"target_language": transcript.target_language,
|
||||
"status": transcript.status,
|
||||
"frontend_url": frontend_url,
|
||||
"action_items": transcript.action_items,
|
||||
},
|
||||
"room": {
|
||||
"id": room.id,
|
||||
|
||||
@@ -318,6 +318,14 @@ async def dummy_storage():
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_settings():
|
||||
"""Provide isolated settings for tests to avoid modifying global settings"""
|
||||
from reflector.settings import Settings
|
||||
|
||||
return Settings()
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def celery_enable_logging():
|
||||
return True
|
||||
|
||||
488
server/tests/test_llm_retry.py
Normal file
488
server/tests/test_llm_retry.py
Normal file
@@ -0,0 +1,488 @@
|
||||
"""Tests for LLM parse error recovery using llama-index Workflow"""
|
||||
|
||||
from time import monotonic
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from pydantic import BaseModel, Field
|
||||
from workflows.errors import WorkflowRuntimeError, WorkflowTimeoutError
|
||||
|
||||
from reflector.llm import LLM, LLMParseError, StructuredOutputWorkflow
|
||||
from reflector.utils.retry import RetryException
|
||||
|
||||
|
||||
class TestResponse(BaseModel):
|
||||
"""Test response model for structured output"""
|
||||
|
||||
title: str = Field(description="A title")
|
||||
summary: str = Field(description="A summary")
|
||||
confidence: float = Field(description="Confidence score", ge=0, le=1)
|
||||
|
||||
|
||||
def make_completion_response(text: str):
|
||||
"""Create a mock CompletionResponse with .text attribute"""
|
||||
response = MagicMock()
|
||||
response.text = text
|
||||
return response
|
||||
|
||||
|
||||
class TestLLMParseErrorRecovery:
|
||||
"""Test parse error recovery with Workflow feedback loop"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_parse_error_recovery_with_feedback(self, test_settings):
|
||||
"""Test that parse errors trigger retry with error feedback"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
# TreeSummarize returns plain text analysis (step 1)
|
||||
mock_summarizer.aget_response = AsyncMock(
|
||||
return_value="The analysis shows a test with summary and high confidence."
|
||||
)
|
||||
|
||||
call_count = {"count": 0}
|
||||
|
||||
async def acomplete_handler(prompt, *args, **kwargs):
|
||||
call_count["count"] += 1
|
||||
if call_count["count"] == 1:
|
||||
# First JSON formatting call returns invalid JSON
|
||||
return make_completion_response('{"title": "Test"}')
|
||||
else:
|
||||
# Second call should have error feedback in prompt
|
||||
assert "Your previous response could not be parsed:" in prompt
|
||||
assert '{"title": "Test"}' in prompt
|
||||
assert "Error:" in prompt
|
||||
assert "Please try again" in prompt
|
||||
return make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.95}'
|
||||
)
|
||||
|
||||
mock_settings.llm.acomplete = AsyncMock(side_effect=acomplete_handler)
|
||||
|
||||
result = await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
assert result.title == "Test"
|
||||
assert result.summary == "Summary"
|
||||
assert result.confidence == 0.95
|
||||
# TreeSummarize called once, Settings.llm.acomplete called twice
|
||||
assert mock_summarizer.aget_response.call_count == 1
|
||||
assert call_count["count"] == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_parse_retry_attempts(self, test_settings):
|
||||
"""Test that parse error retry stops after max attempts"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
# Always return invalid JSON from acomplete
|
||||
mock_settings.llm.acomplete = AsyncMock(
|
||||
return_value=make_completion_response(
|
||||
'{"invalid": "missing required fields"}'
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(LLMParseError, match="Failed to parse"):
|
||||
await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
expected_attempts = test_settings.LLM_PARSE_MAX_RETRIES + 1
|
||||
# TreeSummarize called once, acomplete called max_retries times
|
||||
assert mock_summarizer.aget_response.call_count == 1
|
||||
assert mock_settings.llm.acomplete.call_count == expected_attempts
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_raw_response_logging_on_parse_error(self, test_settings, caplog):
|
||||
"""Test that raw response is logged when parse error occurs"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
caplog.at_level("ERROR"),
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
call_count = {"count": 0}
|
||||
|
||||
async def acomplete_handler(*args, **kwargs):
|
||||
call_count["count"] += 1
|
||||
if call_count["count"] == 1:
|
||||
return make_completion_response('{"title": "Test"}') # Invalid
|
||||
return make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.95}'
|
||||
)
|
||||
|
||||
mock_settings.llm.acomplete = AsyncMock(side_effect=acomplete_handler)
|
||||
|
||||
result = await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
assert result.title == "Test"
|
||||
|
||||
error_logs = [r for r in caplog.records if r.levelname == "ERROR"]
|
||||
raw_response_logged = any("Raw response:" in r.message for r in error_logs)
|
||||
assert raw_response_logged, "Raw response should be logged on parse error"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_validation_errors_in_feedback(self, test_settings):
|
||||
"""Test that validation errors are included in feedback"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
call_count = {"count": 0}
|
||||
|
||||
async def acomplete_handler(prompt, *args, **kwargs):
|
||||
call_count["count"] += 1
|
||||
if call_count["count"] == 1:
|
||||
# Missing title and summary
|
||||
return make_completion_response('{"confidence": 0.5}')
|
||||
else:
|
||||
# Should have schema validation errors in prompt
|
||||
assert (
|
||||
"Schema validation errors" in prompt
|
||||
or "error" in prompt.lower()
|
||||
)
|
||||
return make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.95}'
|
||||
)
|
||||
|
||||
mock_settings.llm.acomplete = AsyncMock(side_effect=acomplete_handler)
|
||||
|
||||
result = await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
assert result.title == "Test"
|
||||
assert call_count["count"] == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_success_on_first_attempt(self, test_settings):
|
||||
"""Test that no retry happens when first attempt succeeds"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
mock_settings.llm.acomplete = AsyncMock(
|
||||
return_value=make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.95}'
|
||||
)
|
||||
)
|
||||
|
||||
result = await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
assert result.title == "Test"
|
||||
assert result.summary == "Summary"
|
||||
assert result.confidence == 0.95
|
||||
assert mock_summarizer.aget_response.call_count == 1
|
||||
assert mock_settings.llm.acomplete.call_count == 1
|
||||
|
||||
|
||||
class TestStructuredOutputWorkflow:
|
||||
"""Direct tests for the StructuredOutputWorkflow"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_workflow_retries_on_validation_error(self):
|
||||
"""Test workflow retries when validation fails"""
|
||||
workflow = StructuredOutputWorkflow(
|
||||
output_cls=TestResponse,
|
||||
max_retries=3,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
call_count = {"count": 0}
|
||||
|
||||
async def acomplete_handler(*args, **kwargs):
|
||||
call_count["count"] += 1
|
||||
if call_count["count"] < 2:
|
||||
return make_completion_response('{"title": "Only title"}')
|
||||
return make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.9}'
|
||||
)
|
||||
|
||||
mock_settings.llm.acomplete = AsyncMock(side_effect=acomplete_handler)
|
||||
|
||||
result = await workflow.run(
|
||||
prompt="Extract data",
|
||||
texts=["Some text"],
|
||||
tone_name=None,
|
||||
)
|
||||
|
||||
assert "success" in result
|
||||
assert result["success"].title == "Test"
|
||||
assert call_count["count"] == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_workflow_returns_error_after_max_retries(self):
|
||||
"""Test workflow returns error after exhausting retries"""
|
||||
workflow = StructuredOutputWorkflow(
|
||||
output_cls=TestResponse,
|
||||
max_retries=2,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
# Always return invalid JSON
|
||||
mock_settings.llm.acomplete = AsyncMock(
|
||||
return_value=make_completion_response('{"invalid": true}')
|
||||
)
|
||||
|
||||
result = await workflow.run(
|
||||
prompt="Extract data",
|
||||
texts=["Some text"],
|
||||
tone_name=None,
|
||||
)
|
||||
|
||||
assert "error" in result
|
||||
# TreeSummarize called once, acomplete called max_retries times
|
||||
assert mock_summarizer.aget_response.call_count == 1
|
||||
assert mock_settings.llm.acomplete.call_count == 2
|
||||
|
||||
|
||||
class TestNetworkErrorRetries:
|
||||
"""Test that network error retries are handled by OpenAILike, not Workflow"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_network_error_propagates_after_openai_retries(self, test_settings):
|
||||
"""Test that network errors are retried by OpenAILike and then propagate.
|
||||
|
||||
Network retries are handled by OpenAILike (max_retries=3), not by our
|
||||
StructuredOutputWorkflow. This test verifies that network errors propagate
|
||||
up after OpenAILike exhausts its retries.
|
||||
"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
# Simulate network error from acomplete (after OpenAILike retries exhausted)
|
||||
network_error = ConnectionError("Connection refused")
|
||||
mock_settings.llm.acomplete = AsyncMock(side_effect=network_error)
|
||||
|
||||
# Network error wrapped in WorkflowRuntimeError
|
||||
with pytest.raises(WorkflowRuntimeError, match="Connection refused"):
|
||||
await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
# acomplete called only once - network error propagates, not retried by Workflow
|
||||
assert mock_settings.llm.acomplete.call_count == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_network_error_not_retried_by_workflow(self, test_settings):
|
||||
"""Test that Workflow does NOT retry network errors (OpenAILike handles those).
|
||||
|
||||
This verifies the separation of concerns:
|
||||
- StructuredOutputWorkflow: retries parse/validation errors
|
||||
- OpenAILike: retries network errors (internally, max_retries=3)
|
||||
"""
|
||||
workflow = StructuredOutputWorkflow(
|
||||
output_cls=TestResponse,
|
||||
max_retries=3,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
with (
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
|
||||
# Network error should propagate immediately, not trigger Workflow retry
|
||||
mock_settings.llm.acomplete = AsyncMock(
|
||||
side_effect=TimeoutError("Request timed out")
|
||||
)
|
||||
|
||||
# Network error wrapped in WorkflowRuntimeError
|
||||
with pytest.raises(WorkflowRuntimeError, match="Request timed out"):
|
||||
await workflow.run(
|
||||
prompt="Extract data",
|
||||
texts=["Some text"],
|
||||
tone_name=None,
|
||||
)
|
||||
|
||||
# Only called once - Workflow doesn't retry network errors
|
||||
assert mock_settings.llm.acomplete.call_count == 1
|
||||
|
||||
|
||||
class TestWorkflowTimeoutRetry:
|
||||
"""Test timeout retry mechanism in get_structured_response"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeout_retry_succeeds_on_retry(self, test_settings):
|
||||
"""Test that WorkflowTimeoutError triggers retry and succeeds"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
call_count = {"count": 0}
|
||||
|
||||
async def workflow_run_side_effect(*args, **kwargs):
|
||||
call_count["count"] += 1
|
||||
if call_count["count"] == 1:
|
||||
raise WorkflowTimeoutError("Operation timed out after 120 seconds")
|
||||
return {
|
||||
"success": TestResponse(
|
||||
title="Test", summary="Summary", confidence=0.95
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
patch("reflector.llm.StructuredOutputWorkflow") as mock_workflow_class,
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_workflow = MagicMock()
|
||||
mock_workflow.run = AsyncMock(side_effect=workflow_run_side_effect)
|
||||
mock_workflow_class.return_value = mock_workflow
|
||||
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
mock_settings.llm.acomplete = AsyncMock(
|
||||
return_value=make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.95}'
|
||||
)
|
||||
)
|
||||
|
||||
result = await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
assert result.title == "Test"
|
||||
assert result.summary == "Summary"
|
||||
assert call_count["count"] == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeout_retry_exhausts_after_max_attempts(self, test_settings):
|
||||
"""Test that timeout retry stops after max attempts"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
call_count = {"count": 0}
|
||||
|
||||
async def workflow_run_side_effect(*args, **kwargs):
|
||||
call_count["count"] += 1
|
||||
raise WorkflowTimeoutError("Operation timed out after 120 seconds")
|
||||
|
||||
with (
|
||||
patch("reflector.llm.StructuredOutputWorkflow") as mock_workflow_class,
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_workflow = MagicMock()
|
||||
mock_workflow.run = AsyncMock(side_effect=workflow_run_side_effect)
|
||||
mock_workflow_class.return_value = mock_workflow
|
||||
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
mock_settings.llm.acomplete = AsyncMock(
|
||||
return_value=make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.95}'
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(RetryException, match="Retry attempts exceeded"):
|
||||
await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
assert call_count["count"] == 3
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_timeout_retry_with_backoff(self, test_settings):
|
||||
"""Test that exponential backoff is applied between retries"""
|
||||
llm = LLM(settings=test_settings, temperature=0.4, max_tokens=100)
|
||||
|
||||
call_times = []
|
||||
|
||||
async def workflow_run_side_effect(*args, **kwargs):
|
||||
call_times.append(monotonic())
|
||||
if len(call_times) < 3:
|
||||
raise WorkflowTimeoutError("Operation timed out after 120 seconds")
|
||||
return {
|
||||
"success": TestResponse(
|
||||
title="Test", summary="Summary", confidence=0.95
|
||||
)
|
||||
}
|
||||
|
||||
with (
|
||||
patch("reflector.llm.StructuredOutputWorkflow") as mock_workflow_class,
|
||||
patch("reflector.llm.TreeSummarize") as mock_summarize,
|
||||
patch("reflector.llm.Settings") as mock_settings,
|
||||
):
|
||||
mock_workflow = MagicMock()
|
||||
mock_workflow.run = AsyncMock(side_effect=workflow_run_side_effect)
|
||||
mock_workflow_class.return_value = mock_workflow
|
||||
|
||||
mock_summarizer = MagicMock()
|
||||
mock_summarize.return_value = mock_summarizer
|
||||
mock_summarizer.aget_response = AsyncMock(return_value="Some analysis")
|
||||
mock_settings.llm.acomplete = AsyncMock(
|
||||
return_value=make_completion_response(
|
||||
'{"title": "Test", "summary": "Summary", "confidence": 0.95}'
|
||||
)
|
||||
)
|
||||
|
||||
result = await llm.get_structured_response(
|
||||
prompt="Test prompt", texts=["Test text"], output_cls=TestResponse
|
||||
)
|
||||
|
||||
assert result.title == "Test"
|
||||
if len(call_times) >= 2:
|
||||
time_between_calls = call_times[1] - call_times[0]
|
||||
assert (
|
||||
time_between_calls >= 1.5
|
||||
), f"Expected ~2s backoff, got {time_between_calls}s"
|
||||
@@ -266,7 +266,11 @@ async def mock_summary_processor():
|
||||
# When flush is called, simulate summary generation by calling the callbacks
|
||||
async def flush_with_callback():
|
||||
mock_summary.flush_called = True
|
||||
from reflector.processors.types import FinalLongSummary, FinalShortSummary
|
||||
from reflector.processors.types import (
|
||||
ActionItems,
|
||||
FinalLongSummary,
|
||||
FinalShortSummary,
|
||||
)
|
||||
|
||||
if hasattr(mock_summary, "_callback"):
|
||||
await mock_summary._callback(
|
||||
@@ -276,12 +280,19 @@ async def mock_summary_processor():
|
||||
await mock_summary._on_short_summary(
|
||||
FinalShortSummary(short_summary="Test short summary", duration=10.0)
|
||||
)
|
||||
if hasattr(mock_summary, "_on_action_items"):
|
||||
await mock_summary._on_action_items(
|
||||
ActionItems(action_items={"test": "action item"})
|
||||
)
|
||||
|
||||
mock_summary.flush = flush_with_callback
|
||||
|
||||
def init_with_callback(transcript=None, callback=None, on_short_summary=None):
|
||||
def init_with_callback(
|
||||
transcript=None, callback=None, on_short_summary=None, on_action_items=None
|
||||
):
|
||||
mock_summary._callback = callback
|
||||
mock_summary._on_short_summary = on_short_summary
|
||||
mock_summary._on_action_items = on_action_items
|
||||
return mock_summary
|
||||
|
||||
mock_summary_class.side_effect = init_with_callback
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import pytest
|
||||
|
||||
from reflector.db.rooms import rooms_controller
|
||||
from reflector.db.transcripts import transcripts_controller
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcript_create(client):
|
||||
@@ -182,3 +185,51 @@ async def test_transcript_mark_reviewed(authenticated_client, client):
|
||||
response = await client.get(f"/transcripts/{tid}")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["reviewed"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcript_get_returns_room_name(authenticated_client, client):
|
||||
"""Test that getting a transcript returns its room_name when linked to a room."""
|
||||
# Create a room
|
||||
room = await rooms_controller.add(
|
||||
name="test-room-for-transcript",
|
||||
user_id="test-user",
|
||||
zulip_auto_post=False,
|
||||
zulip_stream="",
|
||||
zulip_topic="",
|
||||
is_locked=False,
|
||||
room_mode="normal",
|
||||
recording_type="cloud",
|
||||
recording_trigger="automatic-2nd-participant",
|
||||
is_shared=False,
|
||||
webhook_url="",
|
||||
webhook_secret="",
|
||||
)
|
||||
|
||||
# Create a transcript linked to the room
|
||||
transcript = await transcripts_controller.add(
|
||||
name="transcript-with-room",
|
||||
source_kind="file",
|
||||
room_id=room.id,
|
||||
)
|
||||
|
||||
# Get the transcript and verify room_name is returned
|
||||
response = await client.get(f"/transcripts/{transcript.id}")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["room_id"] == room.id
|
||||
assert response.json()["room_name"] == "test-room-for-transcript"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcript_get_returns_null_room_name_when_no_room(
|
||||
authenticated_client, client
|
||||
):
|
||||
"""Test that room_name is null when transcript has no room."""
|
||||
response = await client.post("/transcripts", json={"name": "no-room-transcript"})
|
||||
assert response.status_code == 200
|
||||
tid = response.json()["id"]
|
||||
|
||||
response = await client.get(f"/transcripts/{tid}")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["room_id"] is None
|
||||
assert response.json()["room_name"] is None
|
||||
|
||||
@@ -15,9 +15,12 @@ import {
|
||||
createListCollection,
|
||||
useDisclosure,
|
||||
Tabs,
|
||||
Popover,
|
||||
Text,
|
||||
HStack,
|
||||
} from "@chakra-ui/react";
|
||||
import { useEffect, useMemo, useState } from "react";
|
||||
import { LuEye, LuEyeOff } from "react-icons/lu";
|
||||
import { LuEye, LuEyeOff, LuInfo } from "react-icons/lu";
|
||||
import useRoomList from "./useRoomList";
|
||||
import type { components } from "../../reflector-api";
|
||||
import {
|
||||
@@ -67,6 +70,11 @@ const recordingTypeOptions: SelectOption[] = [
|
||||
{ label: "Cloud", value: "cloud" },
|
||||
];
|
||||
|
||||
const platformOptions: SelectOption[] = [
|
||||
{ label: "Whereby", value: "whereby" },
|
||||
{ label: "Daily", value: "daily" },
|
||||
];
|
||||
|
||||
const roomInitialState = {
|
||||
name: "",
|
||||
zulipAutoPost: false,
|
||||
@@ -82,6 +90,7 @@ const roomInitialState = {
|
||||
icsUrl: "",
|
||||
icsEnabled: false,
|
||||
icsFetchInterval: 5,
|
||||
platform: "whereby",
|
||||
};
|
||||
|
||||
export default function RoomsList() {
|
||||
@@ -99,6 +108,11 @@ export default function RoomsList() {
|
||||
const recordingTypeCollection = createListCollection({
|
||||
items: recordingTypeOptions,
|
||||
});
|
||||
|
||||
const platformCollection = createListCollection({
|
||||
items: platformOptions,
|
||||
});
|
||||
|
||||
const [roomInput, setRoomInput] = useState<null | typeof roomInitialState>(
|
||||
null,
|
||||
);
|
||||
@@ -143,15 +157,24 @@ export default function RoomsList() {
|
||||
zulipStream: detailedEditedRoom.zulip_stream,
|
||||
zulipTopic: detailedEditedRoom.zulip_topic,
|
||||
isLocked: detailedEditedRoom.is_locked,
|
||||
roomMode: detailedEditedRoom.room_mode,
|
||||
roomMode:
|
||||
detailedEditedRoom.platform === "daily"
|
||||
? "group"
|
||||
: detailedEditedRoom.room_mode,
|
||||
recordingType: detailedEditedRoom.recording_type,
|
||||
recordingTrigger: detailedEditedRoom.recording_trigger,
|
||||
recordingTrigger:
|
||||
detailedEditedRoom.platform === "daily"
|
||||
? detailedEditedRoom.recording_type === "cloud"
|
||||
? "automatic-2nd-participant"
|
||||
: "none"
|
||||
: detailedEditedRoom.recording_trigger,
|
||||
isShared: detailedEditedRoom.is_shared,
|
||||
webhookUrl: detailedEditedRoom.webhook_url || "",
|
||||
webhookSecret: detailedEditedRoom.webhook_secret || "",
|
||||
icsUrl: detailedEditedRoom.ics_url || "",
|
||||
icsEnabled: detailedEditedRoom.ics_enabled || false,
|
||||
icsFetchInterval: detailedEditedRoom.ics_fetch_interval || 5,
|
||||
platform: detailedEditedRoom.platform,
|
||||
}
|
||||
: null,
|
||||
[detailedEditedRoom],
|
||||
@@ -277,21 +300,32 @@ export default function RoomsList() {
|
||||
return;
|
||||
}
|
||||
|
||||
const platform: "whereby" | "daily" | null =
|
||||
room.platform === "whereby" || room.platform === "daily"
|
||||
? room.platform
|
||||
: null;
|
||||
|
||||
const roomData = {
|
||||
name: room.name,
|
||||
zulip_auto_post: room.zulipAutoPost,
|
||||
zulip_stream: room.zulipStream,
|
||||
zulip_topic: room.zulipTopic,
|
||||
is_locked: room.isLocked,
|
||||
room_mode: room.roomMode,
|
||||
room_mode: platform === "daily" ? "group" : room.roomMode,
|
||||
recording_type: room.recordingType,
|
||||
recording_trigger: room.recordingTrigger,
|
||||
recording_trigger:
|
||||
platform === "daily"
|
||||
? room.recordingType === "cloud"
|
||||
? "automatic-2nd-participant"
|
||||
: "none"
|
||||
: room.recordingTrigger,
|
||||
is_shared: room.isShared,
|
||||
webhook_url: room.webhookUrl,
|
||||
webhook_secret: room.webhookSecret,
|
||||
ics_url: room.icsUrl,
|
||||
ics_enabled: room.icsEnabled,
|
||||
ics_fetch_interval: room.icsFetchInterval,
|
||||
platform,
|
||||
};
|
||||
|
||||
if (isEditing) {
|
||||
@@ -339,15 +373,21 @@ export default function RoomsList() {
|
||||
zulipStream: roomData.zulip_stream,
|
||||
zulipTopic: roomData.zulip_topic,
|
||||
isLocked: roomData.is_locked,
|
||||
roomMode: roomData.room_mode,
|
||||
roomMode: roomData.platform === "daily" ? "group" : roomData.room_mode, // Daily always uses 2-200
|
||||
recordingType: roomData.recording_type,
|
||||
recordingTrigger: roomData.recording_trigger,
|
||||
recordingTrigger:
|
||||
roomData.platform === "daily"
|
||||
? roomData.recording_type === "cloud"
|
||||
? "automatic-2nd-participant"
|
||||
: "none"
|
||||
: roomData.recording_trigger,
|
||||
isShared: roomData.is_shared,
|
||||
webhookUrl: roomData.webhook_url || "",
|
||||
webhookSecret: roomData.webhook_secret || "",
|
||||
icsUrl: roomData.ics_url || "",
|
||||
icsEnabled: roomData.ics_enabled || false,
|
||||
icsFetchInterval: roomData.ics_fetch_interval || 5,
|
||||
platform: roomData.platform,
|
||||
});
|
||||
setEditRoomId(roomId);
|
||||
setIsEditing(true);
|
||||
@@ -482,6 +522,52 @@ export default function RoomsList() {
|
||||
)}
|
||||
</Field.Root>
|
||||
|
||||
<Field.Root mt={4}>
|
||||
<Field.Label>Platform</Field.Label>
|
||||
<Select.Root
|
||||
value={[room.platform]}
|
||||
onValueChange={(e) => {
|
||||
const newPlatform = e.value[0] as "whereby" | "daily";
|
||||
const updates: Partial<typeof room> = {
|
||||
platform: newPlatform,
|
||||
};
|
||||
if (newPlatform === "daily") {
|
||||
updates.roomMode = "group";
|
||||
updates.recordingTrigger =
|
||||
room.recordingType === "cloud"
|
||||
? "automatic-2nd-participant"
|
||||
: "none";
|
||||
} else {
|
||||
if (room.recordingType !== "cloud") {
|
||||
updates.recordingTrigger = "none";
|
||||
}
|
||||
}
|
||||
setRoomInput({ ...room, ...updates });
|
||||
}}
|
||||
collection={platformCollection}
|
||||
>
|
||||
<Select.HiddenSelect />
|
||||
<Select.Control>
|
||||
<Select.Trigger>
|
||||
<Select.ValueText placeholder="Select platform" />
|
||||
</Select.Trigger>
|
||||
<Select.IndicatorGroup>
|
||||
<Select.Indicator />
|
||||
</Select.IndicatorGroup>
|
||||
</Select.Control>
|
||||
<Select.Positioner>
|
||||
<Select.Content>
|
||||
{platformOptions.map((option) => (
|
||||
<Select.Item key={option.value} item={option}>
|
||||
{option.label}
|
||||
<Select.ItemIndicator />
|
||||
</Select.Item>
|
||||
))}
|
||||
</Select.Content>
|
||||
</Select.Positioner>
|
||||
</Select.Root>
|
||||
</Field.Root>
|
||||
|
||||
<Field.Root mt={4}>
|
||||
<Checkbox.Root
|
||||
name="isLocked"
|
||||
@@ -504,50 +590,95 @@ export default function RoomsList() {
|
||||
<Checkbox.Label>Locked room</Checkbox.Label>
|
||||
</Checkbox.Root>
|
||||
</Field.Root>
|
||||
{room.platform !== "daily" && (
|
||||
<Field.Root mt={4}>
|
||||
<Field.Label>Room size</Field.Label>
|
||||
<Select.Root
|
||||
value={[room.roomMode]}
|
||||
onValueChange={(e) =>
|
||||
setRoomInput({ ...room, roomMode: e.value[0] })
|
||||
}
|
||||
collection={roomModeCollection}
|
||||
>
|
||||
<Select.HiddenSelect />
|
||||
<Select.Control>
|
||||
<Select.Trigger>
|
||||
<Select.ValueText placeholder="Select room size" />
|
||||
</Select.Trigger>
|
||||
<Select.IndicatorGroup>
|
||||
<Select.Indicator />
|
||||
</Select.IndicatorGroup>
|
||||
</Select.Control>
|
||||
<Select.Positioner>
|
||||
<Select.Content>
|
||||
{roomModeOptions.map((option) => (
|
||||
<Select.Item key={option.value} item={option}>
|
||||
{option.label}
|
||||
<Select.ItemIndicator />
|
||||
</Select.Item>
|
||||
))}
|
||||
</Select.Content>
|
||||
</Select.Positioner>
|
||||
</Select.Root>
|
||||
</Field.Root>
|
||||
)}
|
||||
<Field.Root mt={4}>
|
||||
<Field.Label>Room size</Field.Label>
|
||||
<Select.Root
|
||||
value={[room.roomMode]}
|
||||
onValueChange={(e) =>
|
||||
setRoomInput({ ...room, roomMode: e.value[0] })
|
||||
}
|
||||
collection={roomModeCollection}
|
||||
>
|
||||
<Select.HiddenSelect />
|
||||
<Select.Control>
|
||||
<Select.Trigger>
|
||||
<Select.ValueText placeholder="Select room size" />
|
||||
</Select.Trigger>
|
||||
<Select.IndicatorGroup>
|
||||
<Select.Indicator />
|
||||
</Select.IndicatorGroup>
|
||||
</Select.Control>
|
||||
<Select.Positioner>
|
||||
<Select.Content>
|
||||
{roomModeOptions.map((option) => (
|
||||
<Select.Item key={option.value} item={option}>
|
||||
{option.label}
|
||||
<Select.ItemIndicator />
|
||||
</Select.Item>
|
||||
))}
|
||||
</Select.Content>
|
||||
</Select.Positioner>
|
||||
</Select.Root>
|
||||
</Field.Root>
|
||||
<Field.Root mt={4}>
|
||||
<Field.Label>Recording type</Field.Label>
|
||||
<HStack gap={2} alignItems="center">
|
||||
<Field.Label>Recording type</Field.Label>
|
||||
<Popover.Root>
|
||||
<Popover.Trigger asChild>
|
||||
<IconButton
|
||||
aria-label="Recording type help"
|
||||
variant="ghost"
|
||||
size="xs"
|
||||
colorPalette="gray"
|
||||
>
|
||||
<LuInfo />
|
||||
</IconButton>
|
||||
</Popover.Trigger>
|
||||
<Popover.Positioner>
|
||||
<Popover.Content>
|
||||
<Popover.Arrow />
|
||||
<Popover.Body>
|
||||
<Text fontSize="sm" lineHeight="1.6">
|
||||
<strong>None:</strong> No recording will be
|
||||
created.
|
||||
<br />
|
||||
<br />
|
||||
<strong>Local:</strong> Recording happens on
|
||||
each participant's device. Files are saved
|
||||
locally.
|
||||
<br />
|
||||
<br />
|
||||
<strong>Cloud:</strong> Recording happens on
|
||||
the platform's servers and is available after
|
||||
the meeting ends.
|
||||
</Text>
|
||||
</Popover.Body>
|
||||
</Popover.Content>
|
||||
</Popover.Positioner>
|
||||
</Popover.Root>
|
||||
</HStack>
|
||||
<Select.Root
|
||||
value={[room.recordingType]}
|
||||
onValueChange={(e) =>
|
||||
setRoomInput({
|
||||
...room,
|
||||
recordingType: e.value[0],
|
||||
recordingTrigger:
|
||||
e.value[0] !== "cloud"
|
||||
onValueChange={(e) => {
|
||||
const newRecordingType = e.value[0];
|
||||
const updates: Partial<typeof room> = {
|
||||
recordingType: newRecordingType,
|
||||
};
|
||||
if (room.platform === "daily") {
|
||||
updates.recordingTrigger =
|
||||
newRecordingType === "cloud"
|
||||
? "automatic-2nd-participant"
|
||||
: "none";
|
||||
} else {
|
||||
updates.recordingTrigger =
|
||||
newRecordingType !== "cloud"
|
||||
? "none"
|
||||
: room.recordingTrigger,
|
||||
})
|
||||
}
|
||||
: room.recordingTrigger;
|
||||
}
|
||||
setRoomInput({ ...room, ...updates });
|
||||
}}
|
||||
collection={recordingTypeCollection}
|
||||
>
|
||||
<Select.HiddenSelect />
|
||||
@@ -571,40 +702,77 @@ export default function RoomsList() {
|
||||
</Select.Positioner>
|
||||
</Select.Root>
|
||||
</Field.Root>
|
||||
<Field.Root mt={4}>
|
||||
<Field.Label>Cloud recording start trigger</Field.Label>
|
||||
<Select.Root
|
||||
value={[room.recordingTrigger]}
|
||||
onValueChange={(e) =>
|
||||
setRoomInput({
|
||||
...room,
|
||||
recordingTrigger: e.value[0],
|
||||
})
|
||||
}
|
||||
collection={recordingTriggerCollection}
|
||||
disabled={room.recordingType !== "cloud"}
|
||||
>
|
||||
<Select.HiddenSelect />
|
||||
<Select.Control>
|
||||
<Select.Trigger>
|
||||
<Select.ValueText placeholder="Select trigger" />
|
||||
</Select.Trigger>
|
||||
<Select.IndicatorGroup>
|
||||
<Select.Indicator />
|
||||
</Select.IndicatorGroup>
|
||||
</Select.Control>
|
||||
<Select.Positioner>
|
||||
<Select.Content>
|
||||
{recordingTriggerOptions.map((option) => (
|
||||
<Select.Item key={option.value} item={option}>
|
||||
{option.label}
|
||||
<Select.ItemIndicator />
|
||||
</Select.Item>
|
||||
))}
|
||||
</Select.Content>
|
||||
</Select.Positioner>
|
||||
</Select.Root>
|
||||
</Field.Root>
|
||||
{room.recordingType === "cloud" &&
|
||||
room.platform !== "daily" && (
|
||||
<Field.Root mt={4}>
|
||||
<HStack gap={2} alignItems="center">
|
||||
<Field.Label>Recording start trigger</Field.Label>
|
||||
<Popover.Root>
|
||||
<Popover.Trigger asChild>
|
||||
<IconButton
|
||||
aria-label="Recording start trigger help"
|
||||
variant="ghost"
|
||||
size="xs"
|
||||
colorPalette="gray"
|
||||
>
|
||||
<LuInfo />
|
||||
</IconButton>
|
||||
</Popover.Trigger>
|
||||
<Popover.Positioner>
|
||||
<Popover.Content>
|
||||
<Popover.Arrow />
|
||||
<Popover.Body>
|
||||
<Text fontSize="sm" lineHeight="1.6">
|
||||
<strong>None:</strong> Recording must be
|
||||
started manually by a participant.
|
||||
<br />
|
||||
<br />
|
||||
<strong>Prompt:</strong> Participants will
|
||||
be prompted to start recording when they
|
||||
join.
|
||||
<br />
|
||||
<br />
|
||||
<strong>Automatic:</strong> Recording
|
||||
starts automatically when a second
|
||||
participant joins.
|
||||
</Text>
|
||||
</Popover.Body>
|
||||
</Popover.Content>
|
||||
</Popover.Positioner>
|
||||
</Popover.Root>
|
||||
</HStack>
|
||||
<Select.Root
|
||||
value={[room.recordingTrigger]}
|
||||
onValueChange={(e) =>
|
||||
setRoomInput({
|
||||
...room,
|
||||
recordingTrigger: e.value[0],
|
||||
})
|
||||
}
|
||||
collection={recordingTriggerCollection}
|
||||
>
|
||||
<Select.HiddenSelect />
|
||||
<Select.Control>
|
||||
<Select.Trigger>
|
||||
<Select.ValueText placeholder="Select trigger" />
|
||||
</Select.Trigger>
|
||||
<Select.IndicatorGroup>
|
||||
<Select.Indicator />
|
||||
</Select.IndicatorGroup>
|
||||
</Select.Control>
|
||||
<Select.Positioner>
|
||||
<Select.Content>
|
||||
{recordingTriggerOptions.map((option) => (
|
||||
<Select.Item key={option.value} item={option}>
|
||||
{option.label}
|
||||
<Select.ItemIndicator />
|
||||
</Select.Item>
|
||||
))}
|
||||
</Select.Content>
|
||||
</Select.Positioner>
|
||||
</Select.Root>
|
||||
</Field.Root>
|
||||
)}
|
||||
|
||||
<Field.Root mt={4}>
|
||||
<Checkbox.Root
|
||||
|
||||
@@ -117,15 +117,6 @@ export default function TranscriptDetails(details: TranscriptDetails) {
|
||||
return <Modal title="Loading" text={"Loading transcript..."} />;
|
||||
}
|
||||
|
||||
if (mp3.error) {
|
||||
return (
|
||||
<Modal
|
||||
title="Transcription error"
|
||||
text={`There was an error loading the recording. Error: ${mp3.error}`}
|
||||
/>
|
||||
);
|
||||
}
|
||||
|
||||
return (
|
||||
<>
|
||||
<Grid
|
||||
@@ -147,7 +138,12 @@ export default function TranscriptDetails(details: TranscriptDetails) {
|
||||
/>
|
||||
) : !mp3.loading && (waveform.error || mp3.error) ? (
|
||||
<Box p={4} bg="red.100" borderRadius="md">
|
||||
<Text>Error loading this recording</Text>
|
||||
<Text>
|
||||
Error loading{" "}
|
||||
{[waveform.error && "waveform", mp3.error && "mp3"]
|
||||
.filter(Boolean)
|
||||
.join(" and ")}
|
||||
</Text>
|
||||
</Box>
|
||||
) : (
|
||||
<Skeleton h={14} />
|
||||
|
||||
@@ -2,20 +2,29 @@
|
||||
|
||||
import { Spinner, Link } from "@chakra-ui/react";
|
||||
import { useAuth } from "../lib/AuthProvider";
|
||||
import { usePathname } from "next/navigation";
|
||||
import { getLogoutRedirectUrl } from "../lib/auth";
|
||||
|
||||
export default function UserInfo() {
|
||||
const auth = useAuth();
|
||||
const pathname = usePathname();
|
||||
const status = auth.status;
|
||||
const isLoading = status === "loading";
|
||||
const isAuthenticated = status === "authenticated";
|
||||
const isRefreshing = status === "refreshing";
|
||||
|
||||
const callbackUrl = getLogoutRedirectUrl(pathname);
|
||||
|
||||
return isLoading ? (
|
||||
<Spinner size="xs" className="mx-3" />
|
||||
) : !isAuthenticated && !isRefreshing ? (
|
||||
<Link
|
||||
href="/"
|
||||
href="#"
|
||||
className="font-light px-2"
|
||||
onClick={() => auth.signIn("authentik")}
|
||||
onClick={(e) => {
|
||||
e.preventDefault();
|
||||
auth.signIn("authentik");
|
||||
}}
|
||||
>
|
||||
Log in
|
||||
</Link>
|
||||
@@ -23,7 +32,7 @@ export default function UserInfo() {
|
||||
<Link
|
||||
href="#"
|
||||
className="font-light px-2"
|
||||
onClick={() => auth.signOut({ callbackUrl: "/" })}
|
||||
onClick={() => auth.signOut({ callbackUrl })}
|
||||
>
|
||||
Log out
|
||||
</Link>
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
recordingTypeRequiresConsent,
|
||||
} from "../../lib/consent";
|
||||
import { useRoomJoinMeeting } from "../../lib/apiHooks";
|
||||
import { assertExists } from "../../lib/utils";
|
||||
|
||||
type Meeting = components["schemas"]["Meeting"];
|
||||
|
||||
@@ -22,16 +23,15 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {
|
||||
const router = useRouter();
|
||||
const params = useParams();
|
||||
const auth = useAuth();
|
||||
const status = auth.status;
|
||||
const authLastUserId = auth.lastUserId;
|
||||
const containerRef = useRef<HTMLDivElement>(null);
|
||||
const joinMutation = useRoomJoinMeeting();
|
||||
const [joinedMeeting, setJoinedMeeting] = useState<Meeting | null>(null);
|
||||
|
||||
const roomName = params?.roomName as string;
|
||||
|
||||
// Always call /join to get a fresh token with user_id
|
||||
useEffect(() => {
|
||||
if (status === "loading" || !meeting?.id || !roomName) return;
|
||||
if (authLastUserId === undefined || !meeting?.id || !roomName) return;
|
||||
|
||||
const join = async () => {
|
||||
try {
|
||||
@@ -50,18 +50,17 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {
|
||||
};
|
||||
|
||||
join();
|
||||
}, [meeting?.id, roomName, status]);
|
||||
}, [meeting?.id, roomName, authLastUserId]);
|
||||
|
||||
const roomUrl = joinedMeeting?.host_room_url || joinedMeeting?.room_url;
|
||||
const isLoading =
|
||||
status === "loading" || joinMutation.isPending || !joinedMeeting;
|
||||
const roomUrl = joinedMeeting?.room_url;
|
||||
|
||||
const handleLeave = useCallback(() => {
|
||||
router.push("/browse");
|
||||
}, [router]);
|
||||
|
||||
useEffect(() => {
|
||||
if (isLoading || !roomUrl || !containerRef.current) return;
|
||||
if (authLastUserId === undefined || !roomUrl || !containerRef.current)
|
||||
return;
|
||||
|
||||
let frame: DailyCall | null = null;
|
||||
let destroyed = false;
|
||||
@@ -92,19 +91,41 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {
|
||||
|
||||
frame.on("joined-meeting", async () => {
|
||||
try {
|
||||
await frame.startRecording({ type: "raw-tracks" });
|
||||
const frameInstance = assertExists(
|
||||
frame,
|
||||
"frame object got lost somewhere after frame.on was called",
|
||||
);
|
||||
|
||||
if (meeting.recording_type === "cloud") {
|
||||
console.log("Starting cloud recording");
|
||||
await frameInstance.startRecording({ type: "raw-tracks" });
|
||||
}
|
||||
} catch (error) {
|
||||
console.error("Failed to start recording:", error);
|
||||
}
|
||||
});
|
||||
|
||||
await frame.join({ url: roomUrl });
|
||||
await frame.join({
|
||||
url: roomUrl,
|
||||
sendSettings: {
|
||||
video: {
|
||||
// Optimize bandwidth for camera video
|
||||
// allowAdaptiveLayers automatically adjusts quality based on network conditions
|
||||
allowAdaptiveLayers: true,
|
||||
// Use bandwidth-optimized preset as fallback for browsers without adaptive support
|
||||
maxQuality: "medium",
|
||||
},
|
||||
// Note: screenVideo intentionally not configured to preserve full quality for screen shares
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
console.error("Error creating Daily frame:", error);
|
||||
}
|
||||
};
|
||||
|
||||
createAndJoin();
|
||||
createAndJoin().catch((error) => {
|
||||
console.error("Failed to create and join meeting:", error);
|
||||
});
|
||||
|
||||
return () => {
|
||||
destroyed = true;
|
||||
@@ -114,9 +135,9 @@ export default function DailyRoom({ meeting }: DailyRoomProps) {
|
||||
});
|
||||
}
|
||||
};
|
||||
}, [roomUrl, isLoading, handleLeave]);
|
||||
}, [roomUrl, authLastUserId, handleLeave]);
|
||||
|
||||
if (isLoading) {
|
||||
if (authLastUserId === undefined) {
|
||||
return (
|
||||
<Center width="100vw" height="100vh">
|
||||
<Spinner size="xl" />
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"use client";
|
||||
|
||||
import { createContext, useContext } from "react";
|
||||
import { createContext, useContext, useRef } from "react";
|
||||
import { useSession as useNextAuthSession } from "next-auth/react";
|
||||
import { signOut, signIn } from "next-auth/react";
|
||||
import { configureApiAuth } from "./apiClient";
|
||||
@@ -25,6 +25,9 @@ type AuthContextType = (
|
||||
update: () => Promise<Session | null>;
|
||||
signIn: typeof signIn;
|
||||
signOut: typeof signOut;
|
||||
// TODO probably rename isLoading to isReloading and make THIS field "isLoading"
|
||||
// undefined is "not known", null is "is certainly logged out"
|
||||
lastUserId: CustomSession["user"]["id"] | null | undefined;
|
||||
};
|
||||
|
||||
const AuthContext = createContext<AuthContextType | undefined>(undefined);
|
||||
@@ -41,10 +44,15 @@ const noopAuthContext: AuthContextType = {
|
||||
signOut: async () => {
|
||||
throw new Error("signOut not supposed to be called");
|
||||
},
|
||||
lastUserId: undefined,
|
||||
};
|
||||
|
||||
export function AuthProvider({ children }: { children: React.ReactNode }) {
|
||||
const { data: session, status, update } = useNextAuthSession();
|
||||
// referential comparison done in component, must be primitive /or cached
|
||||
const lastUserId = useRef<CustomSession["user"]["id"] | null | undefined>(
|
||||
null,
|
||||
);
|
||||
|
||||
const contextValue: AuthContextType = isAuthEnabled
|
||||
? {
|
||||
@@ -73,11 +81,16 @@ export function AuthProvider({ children }: { children: React.ReactNode }) {
|
||||
case "authenticated": {
|
||||
const customSession = assertCustomSession(session);
|
||||
if (customSession?.error === REFRESH_ACCESS_TOKEN_ERROR) {
|
||||
// warning: call order-dependent
|
||||
lastUserId.current = null;
|
||||
// token had expired but next auth still returns "authenticated" so show user unauthenticated state
|
||||
return {
|
||||
status: "unauthenticated" as const,
|
||||
};
|
||||
} else if (customSession?.accessToken) {
|
||||
// updates anyways with updated properties below
|
||||
// warning! execution order conscience, must be ran before reading lastUserId.current below
|
||||
lastUserId.current = customSession.user.id;
|
||||
return {
|
||||
status,
|
||||
accessToken: customSession.accessToken,
|
||||
@@ -92,6 +105,8 @@ export function AuthProvider({ children }: { children: React.ReactNode }) {
|
||||
}
|
||||
}
|
||||
case "unauthenticated": {
|
||||
// warning: call order-dependent
|
||||
lastUserId.current = null;
|
||||
return { status: "unauthenticated" as const };
|
||||
}
|
||||
default: {
|
||||
@@ -103,6 +118,8 @@ export function AuthProvider({ children }: { children: React.ReactNode }) {
|
||||
update,
|
||||
signIn,
|
||||
signOut,
|
||||
// for optimistic cases when we assume "loading" doesn't immediately invalidate the user
|
||||
lastUserId: lastUserId.current,
|
||||
}
|
||||
: noopAuthContext;
|
||||
|
||||
|
||||
@@ -18,3 +18,8 @@ export const LOGIN_REQUIRED_PAGES = [
|
||||
export const PROTECTED_PAGES = new RegExp(
|
||||
LOGIN_REQUIRED_PAGES.map((page) => `^${page}$`).join("|"),
|
||||
);
|
||||
|
||||
export function getLogoutRedirectUrl(pathname: string): string {
|
||||
const transcriptPagePattern = /^\/transcripts\/[^/]+$/;
|
||||
return transcriptPagePattern.test(pathname) ? pathname : "/";
|
||||
}
|
||||
|
||||
@@ -148,7 +148,7 @@ export const authOptions = (): AuthOptions =>
|
||||
},
|
||||
async session({ session, token }) {
|
||||
const extendedToken = token as JWTWithAccessToken;
|
||||
|
||||
console.log("extendedToken", extendedToken);
|
||||
const userId = await getUserId(extendedToken.accessToken);
|
||||
|
||||
return {
|
||||
|
||||
@@ -31,7 +31,7 @@
|
||||
"ioredis": "^5.7.0",
|
||||
"jest-worker": "^29.6.2",
|
||||
"lucide-react": "^0.525.0",
|
||||
"next": "^15.5.3",
|
||||
"next": "^15.5.9",
|
||||
"next-auth": "^4.24.7",
|
||||
"next-themes": "^0.4.6",
|
||||
"nuqs": "^2.4.3",
|
||||
|
||||
508
www/pnpm-lock.yaml
generated
508
www/pnpm-lock.yaml
generated
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user