ФУНКЦИИ - МЕЖПРОЕКТНАЯ КОММУНИКАЦИЯ: мониторинг очереди Voice Tasker

plane-src/apps/api/plane/app/urls/voice_tasker.py

@@ -6,6 +6,7 @@ from django.urls import path
 
 from plane.app.views import (
     VoiceTaskCommitEndpoint,
+    VoiceTaskMonitorEndpoint,
     VoiceTaskParseEndpoint,
     VoiceTaskPreflightEndpoint,
     VoiceTaskSessionEndpoint,
@@ -25,6 +26,11 @@ urlpatterns = [
         WorkspaceAISettingsTestConnectionEndpoint.as_view(),
         name="voice-tasker-settings-test-connection",
     ),
+    path(
+        "workspaces/<str:slug>/voice-tasker/monitor/",
+        VoiceTaskMonitorEndpoint.as_view(),
+        name="voice-tasker-monitor",
+    ),
     path(
         "workspaces/<str:slug>/voice-task/preflight/",
         VoiceTaskPreflightEndpoint.as_view(),

plane-src/apps/api/plane/app/views/__init__.py

@@ -251,6 +251,7 @@ from .webhook.base import (
 
 from .voice_tasker import (
     VoiceTaskCommitEndpoint,
+    VoiceTaskMonitorEndpoint,
     VoiceTaskParseEndpoint,
     VoiceTaskPreflightEndpoint,
     VoiceTaskSessionEndpoint,

plane-src/apps/api/plane/app/views/voice_tasker.py

@@ -16,7 +16,7 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 from django.core.files.base import ContentFile
 from django.core.serializers.json import DjangoJSONEncoder
 from django.db import connection, transaction
-from django.db.models import Max, Q
+from django.db.models import Avg, Count, Max, Q, Sum
 from django.utils.text import get_valid_filename
 from django.utils import timezone
 
@@ -68,6 +68,8 @@ VOICE_TASK_RATE_LIMIT_HOURLY_WINDOW_SECONDS = 60 * 60
 VOICE_TASK_RATE_LIMIT_DAILY_WINDOW_SECONDS = 24 * 60 * 60
 VOICE_TASK_CONCURRENCY_RETRY_AFTER_SECONDS = 15
 VOICE_TASK_ACTIVE_SESSION_STALE_SECONDS = 30 * 60
+VOICE_TASK_MONITOR_WINDOW_SECONDS = 24 * 60 * 60
+VOICE_TASK_MONITOR_RECENT_LIMIT = 20
 VOICE_TASK_RATE_LIMIT_ERROR_CODES = {
     "voice_task_user_hourly_limit_exceeded",
     "voice_task_workspace_hourly_limit_exceeded",
@@ -531,6 +533,14 @@ def clear_voice_task_audio_file(voice_session):
     voice_session.save(update_fields=["audio_file", "updated_at"])
 
 
+def get_voice_task_duration_ms(started_at, finished_at=None):
+    if not started_at:
+        return None
+
+    finished_at = finished_at or timezone.now()
+    return max(0, int((finished_at - started_at).total_seconds() * 1000))
+
+
 def get_voice_task_concurrency_state(workspace, ai_settings):
     active_sessions = VoiceTaskSession.objects.filter(
         workspace=workspace,
@@ -578,6 +588,7 @@ def create_voice_task_rate_limit_session(
         user=user,
         project=project,
         status=VoiceTaskSession.Status.FAILED,
+        failed_at=timezone.now(),
         audio_duration_seconds=duration_seconds,
         audio_content_type=audio_content_type,
         audio_size=getattr(audio, "size", None),
@@ -602,6 +613,7 @@ def create_voice_task_failed_preflight_session(
         user=user,
         project=project,
         status=VoiceTaskSession.Status.FAILED,
+        failed_at=timezone.now(),
         audio_duration_seconds=duration_seconds,
         audio_content_type=audio_content_type,
         audio_size=getattr(audio, "size", None),
@@ -666,6 +678,7 @@ def reserve_voice_task_session(
             user=user,
             project=quota_project,
             status=VoiceTaskSession.Status.QUEUED,
+            queued_at=timezone.now(),
             audio_duration_seconds=duration_seconds,
             audio_content_type=audio_content_type,
             audio_size=getattr(audio, "size", None),
@@ -772,6 +785,7 @@ class VoiceTaskParserService:
     def __init__(self, api_key, model):
         self.client = OpenAI(api_key=api_key)
         self.model = model
+        self.last_usage = {}
 
     def parse(self, parser_context):
         response = self.client.chat.completions.create(
@@ -822,6 +836,13 @@ class VoiceTaskParserService:
             ],
         )
         content = response.choices[0].message.content or ""
+        usage = getattr(response, "usage", None)
+        if usage:
+            self.last_usage = {
+                "prompt_tokens": getattr(usage, "prompt_tokens", None),
+                "completion_tokens": getattr(usage, "completion_tokens", None),
+                "total_tokens": getattr(usage, "total_tokens", None),
+            }
         try:
             parsed = json.loads(content)
         except json.JSONDecodeError as exc:
@@ -2671,11 +2692,13 @@ def process_voice_task_session_pipeline(voice_session_id):
         ai_settings, api_key = get_workspace_ai_runtime(workspace)
 
         voice_session.status = VoiceTaskSession.Status.PROCESSING
-        voice_session.save(update_fields=["status", "updated_at"])
+        voice_session.processing_started_at = timezone.now()
+        voice_session.save(update_fields=["status", "processing_started_at", "updated_at"])
 
         voice_session.status = VoiceTaskSession.Status.TRANSCRIBING
         voice_session.save(update_fields=["status", "updated_at"])
 
+        transcription_started_at = timezone.now()
         voice_session.audio_file.open("rb")
         try:
             transcript = OpenAITranscriptionService(
@@ -2698,7 +2721,20 @@ def process_voice_task_session_pipeline(voice_session_id):
 
         voice_session.status = VoiceTaskSession.Status.TRANSCRIBED
         voice_session.transcript = transcript
-        voice_session.save(update_fields=["status", "transcript", "updated_at"])
+        voice_session.transcribed_at = timezone.now()
+        voice_session.transcription_duration_ms = get_voice_task_duration_ms(
+            transcription_started_at,
+            voice_session.transcribed_at,
+        )
+        voice_session.save(
+            update_fields=[
+                "status",
+                "transcript",
+                "transcribed_at",
+                "transcription_duration_ms",
+                "updated_at",
+            ]
+        )
 
         parser_context = build_voice_task_parser_context(
             workspace=workspace,
@@ -2710,38 +2746,93 @@ def process_voice_task_session_pipeline(voice_session_id):
         voice_session.status = VoiceTaskSession.Status.PARSING
         voice_session.save(update_fields=["status", "updated_at"])
 
-        parsed = VoiceTaskParserService(
+        parsing_started_at = timezone.now()
+        parser_service = VoiceTaskParserService(
             api_key=api_key,
             model=ai_settings.structuring_model,
-        ).parse(parser_context)
+        )
+        parsed = parser_service.parse(parser_context)
+        parsing_finished_at = timezone.now()
         if not parsed.get("state_hint"):
             inferred_state_hint = infer_voice_task_state_hint(transcript)
             if inferred_state_hint:
                 parsed["state_hint"] = inferred_state_hint
         parsed = harden_voice_task_intent(parsed, transcript)
+        parser_usage = parser_service.last_usage or {}
 
         voice_session.status = VoiceTaskSession.Status.PARSED
         voice_session.intent = parsed["intent"]
         voice_session.parsed_json = parsed
+        voice_session.completed_at = timezone.now()
+        voice_session.parsing_duration_ms = get_voice_task_duration_ms(parsing_started_at, parsing_finished_at)
+        voice_session.processing_duration_ms = get_voice_task_duration_ms(
+            voice_session.processing_started_at,
+            voice_session.completed_at,
+        )
+        voice_session.parser_prompt_tokens = parser_usage.get("prompt_tokens")
+        voice_session.parser_completion_tokens = parser_usage.get("completion_tokens")
+        voice_session.parser_total_tokens = parser_usage.get("total_tokens")
         if not voice_session.project_id and parsed.get("project_id"):
             voice_session.project = get_voice_task_quota_project(workspace, project_id=parsed.get("project_id"))
         voice_session.error_code = ""
         voice_session.error_message = ""
         voice_session.save(
-            update_fields=["status", "intent", "parsed_json", "project", "error_code", "error_message", "updated_at"]
+            update_fields=[
+                "status",
+                "intent",
+                "parsed_json",
+                "project",
+                "completed_at",
+                "parsing_duration_ms",
+                "processing_duration_ms",
+                "parser_prompt_tokens",
+                "parser_completion_tokens",
+                "parser_total_tokens",
+                "error_code",
+                "error_message",
+                "updated_at",
+            ]
         )
     except VoiceTaskerPipelineError as exc:
         pipeline_error = exc
         voice_session.status = VoiceTaskSession.Status.FAILED
+        voice_session.failed_at = timezone.now()
+        voice_session.processing_duration_ms = get_voice_task_duration_ms(
+            voice_session.processing_started_at,
+            voice_session.failed_at,
+        )
         voice_session.error_code = pipeline_error.code
         voice_session.error_message = pipeline_error.message
-        voice_session.save(update_fields=["status", "error_code", "error_message", "updated_at"])
+        voice_session.save(
+            update_fields=[
+                "status",
+                "failed_at",
+                "processing_duration_ms",
+                "error_code",
+                "error_message",
+                "updated_at",
+            ]
+        )
     except Exception as exc:
         pipeline_error = get_openai_pipeline_error(exc)
         voice_session.status = VoiceTaskSession.Status.FAILED
+        voice_session.failed_at = timezone.now()
+        voice_session.processing_duration_ms = get_voice_task_duration_ms(
+            voice_session.processing_started_at,
+            voice_session.failed_at,
+        )
         voice_session.error_code = pipeline_error.code
         voice_session.error_message = pipeline_error.message
-        voice_session.save(update_fields=["status", "error_code", "error_message", "updated_at"])
+        voice_session.save(
+            update_fields=[
+                "status",
+                "failed_at",
+                "processing_duration_ms",
+                "error_code",
+                "error_message",
+                "updated_at",
+            ]
+        )
     finally:
         try:
             clear_voice_task_audio_file(voice_session)
@@ -2808,6 +2899,211 @@ def serialize_voice_task_session_response(voice_session):
     return base_payload
 
 
+def get_voice_task_stale_threshold():
+    return timezone.now() - timedelta(seconds=VOICE_TASK_ACTIVE_SESSION_STALE_SECONDS)
+
+
+def get_voice_task_user_display_name(user):
+    if not user:
+        return "Пользователь"
+
+    return getattr(user, "display_name", "") or getattr(user, "email", "") or "Пользователь"
+
+
+def serialize_voice_task_monitor_session(voice_session, stale_threshold=None):
+    stale_threshold = stale_threshold or get_voice_task_stale_threshold()
+    user = voice_session.user
+    project = voice_session.project
+    created_issue = voice_session.created_task
+    updated_issue = voice_session.updated_task
+    issue = created_issue or updated_issue
+    is_active = voice_session.status in VOICE_TASK_ACTIVE_SESSION_STATUSES
+
+    return {
+        "id": str(voice_session.id),
+        "status": voice_session.status,
+        "intent": voice_session.intent,
+        "is_active": is_active,
+        "is_stale": bool(is_active and voice_session.updated_at < stale_threshold),
+        "user": {
+            "id": str(user.id) if user else None,
+            "name": get_voice_task_user_display_name(user),
+            "email": getattr(user, "email", "") if user else "",
+        },
+        "project": {
+            "id": str(project.id) if project else None,
+            "name": project.name if project else "",
+            "identifier": project.identifier if project else "",
+        },
+        "issue": {
+            "id": str(issue.id) if issue else None,
+            "key": f"{issue.project.identifier}-{issue.sequence_id}" if issue and issue.project_id else "",
+            "name": issue.name if issue else "",
+        },
+        "audio": {
+            "duration_seconds": voice_session.audio_duration_seconds,
+            "size": voice_session.audio_size,
+            "content_type": voice_session.audio_content_type,
+            "has_temp_file": bool(voice_session.audio_file),
+        },
+        "timings": {
+            "queued_at": voice_session.queued_at,
+            "processing_started_at": voice_session.processing_started_at,
+            "transcribed_at": voice_session.transcribed_at,
+            "completed_at": voice_session.completed_at,
+            "failed_at": voice_session.failed_at,
+            "transcription_duration_ms": voice_session.transcription_duration_ms,
+            "parsing_duration_ms": voice_session.parsing_duration_ms,
+            "processing_duration_ms": voice_session.processing_duration_ms,
+            "last_update_at": voice_session.updated_at,
+            "created_at": voice_session.created_at,
+        },
+        "usage": {
+            "parser_prompt_tokens": voice_session.parser_prompt_tokens,
+            "parser_completion_tokens": voice_session.parser_completion_tokens,
+            "parser_total_tokens": voice_session.parser_total_tokens,
+        },
+        "error": {
+            "code": voice_session.error_code,
+            "message": voice_session.error_message,
+        },
+    }
+
+
+def get_voice_task_monitor_payload(workspace):
+    now = timezone.now()
+    window_start = now - timedelta(seconds=VOICE_TASK_MONITOR_WINDOW_SECONDS)
+    stale_threshold = get_voice_task_stale_threshold()
+    ai_settings, _ = WorkspaceAISettings.objects.get_or_create(workspace=workspace)
+
+    base_sessions = VoiceTaskSession.objects.filter(workspace=workspace)
+    window_sessions = base_sessions.filter(created_at__gte=window_start)
+    active_sessions = base_sessions.filter(
+        status__in=VOICE_TASK_ACTIVE_SESSION_STATUSES,
+        updated_at__gte=stale_threshold,
+    )
+    stale_sessions = base_sessions.filter(
+        status__in=VOICE_TASK_ACTIVE_SESSION_STATUSES,
+        updated_at__lt=stale_threshold,
+    )
+
+    status_counts = {
+        item["status"]: item["count"]
+        for item in window_sessions.values("status").annotate(count=Count("id"))
+    }
+    aggregate = window_sessions.aggregate(
+        total=Count("id"),
+        avg_processing_duration_ms=Avg("processing_duration_ms"),
+        avg_transcription_duration_ms=Avg("transcription_duration_ms"),
+        avg_parsing_duration_ms=Avg("parsing_duration_ms"),
+        total_audio_seconds=Sum("audio_duration_seconds"),
+        total_audio_size=Sum("audio_size"),
+        parser_total_tokens=Sum("parser_total_tokens"),
+    )
+    error_counts = list(
+        window_sessions.filter(status=VoiceTaskSession.Status.FAILED)
+        .exclude(error_code="")
+        .values("error_code")
+        .annotate(count=Count("id"))
+        .order_by("-count")[:8]
+    )
+
+    recent_sessions = (
+        base_sessions.select_related("user", "project", "created_task__project", "updated_task__project")
+        .order_by("-created_at")[:VOICE_TASK_MONITOR_RECENT_LIMIT]
+    )
+    active_session_rows = (
+        base_sessions.select_related("user", "project", "created_task__project", "updated_task__project")
+        .filter(status__in=VOICE_TASK_ACTIVE_SESSION_STATUSES)
+        .order_by("-created_at")[:VOICE_TASK_MONITOR_RECENT_LIMIT]
+    )
+
+    return {
+        "ok": True,
+        "generated_at": now,
+        "window_seconds": VOICE_TASK_MONITOR_WINDOW_SECONDS,
+        "stale_after_seconds": VOICE_TASK_ACTIVE_SESSION_STALE_SECONDS,
+        "concurrency": get_voice_task_concurrency_state(workspace, ai_settings),
+        "summary": {
+            "total": aggregate["total"] or 0,
+            "parsed": status_counts.get(VoiceTaskSession.Status.PARSED, 0),
+            "failed": status_counts.get(VoiceTaskSession.Status.FAILED, 0),
+            "active": active_sessions.count(),
+            "stale": stale_sessions.count(),
+            "avg_processing_duration_ms": aggregate["avg_processing_duration_ms"],
+            "avg_transcription_duration_ms": aggregate["avg_transcription_duration_ms"],
+            "avg_parsing_duration_ms": aggregate["avg_parsing_duration_ms"],
+            "total_audio_seconds": aggregate["total_audio_seconds"] or 0,
+            "total_audio_size": aggregate["total_audio_size"] or 0,
+            "parser_total_tokens": aggregate["parser_total_tokens"] or 0,
+            "status_counts": status_counts,
+            "error_counts": error_counts,
+        },
+        "active_sessions": [
+            serialize_voice_task_monitor_session(session, stale_threshold=stale_threshold)
+            for session in active_session_rows
+        ],
+        "recent_sessions": [
+            serialize_voice_task_monitor_session(session, stale_threshold=stale_threshold)
+            for session in recent_sessions
+        ],
+    }
+
+
+class VoiceTaskMonitorEndpoint(BaseAPIView):
+    @allow_permission(allowed_roles=[ROLE.ADMIN], level="WORKSPACE")
+    def get(self, request, slug):
+        workspace = Workspace.objects.get(slug=slug)
+        return Response(get_voice_task_monitor_payload(workspace), status=status.HTTP_200_OK)
+
+    @allow_permission(allowed_roles=[ROLE.ADMIN], level="WORKSPACE")
+    def post(self, request, slug):
+        workspace = Workspace.objects.get(slug=slug)
+        action = request.data.get("action")
+        if action != "fail_stale":
+            return Response(
+                {"ok": False, "code": "unsupported_action", "error": "Unsupported Voice Task monitor action."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        now = timezone.now()
+        stale_sessions = list(
+            VoiceTaskSession.objects.filter(
+                workspace=workspace,
+                status__in=VOICE_TASK_ACTIVE_SESSION_STATUSES,
+                updated_at__lt=now - timedelta(seconds=VOICE_TASK_ACTIVE_SESSION_STALE_SECONDS),
+            )
+        )
+
+        for voice_session in stale_sessions:
+            voice_session.status = VoiceTaskSession.Status.FAILED
+            voice_session.failed_at = now
+            voice_session.processing_duration_ms = get_voice_task_duration_ms(
+                voice_session.processing_started_at,
+                now,
+            )
+            voice_session.error_code = "voice_task_stale_session"
+            voice_session.error_message = "Voice Tasker session was marked as stale by workspace admin."
+            voice_session.save(
+                update_fields=[
+                    "status",
+                    "failed_at",
+                    "processing_duration_ms",
+                    "error_code",
+                    "error_message",
+                    "updated_at",
+                ]
+            )
+            try:
+                clear_voice_task_audio_file(voice_session)
+            except Exception as exc:
+                log_exception(exc)
+
+        payload = get_voice_task_monitor_payload(workspace)
+        payload["cleaned_count"] = len(stale_sessions)
+        return Response(payload, status=status.HTTP_200_OK)
+
+
 class WorkspaceAISettingsEndpoint(BaseAPIView):
     def get_settings(self, slug):
         workspace = Workspace.objects.get(slug=slug)
@@ -3012,9 +3308,10 @@ class VoiceTaskParseEndpoint(BaseAPIView):
         except Exception as exc:
             log_exception(exc)
             voice_session.status = VoiceTaskSession.Status.FAILED
+            voice_session.failed_at = timezone.now()
             voice_session.error_code = "voice_task_queue_unavailable"
             voice_session.error_message = "Voice Tasker queue is not available."
-            voice_session.save(update_fields=["status", "error_code", "error_message", "updated_at"])
+            voice_session.save(update_fields=["status", "failed_at", "error_code", "error_message", "updated_at"])
             clear_voice_task_audio_file(voice_session)
             return Response(
                 {

plane-src/apps/api/plane/db/migrations/0133_voice_tasker_observability.py

@@ -0,0 +1,68 @@
+# Generated by Codex on 2026-04-28
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("db", "0132_voice_tasker_queue_concurrency"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="completed_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="failed_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="parser_completion_tokens",
+            field=models.PositiveIntegerField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="parser_prompt_tokens",
+            field=models.PositiveIntegerField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="parser_total_tokens",
+            field=models.PositiveIntegerField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="parsing_duration_ms",
+            field=models.PositiveIntegerField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="processing_duration_ms",
+            field=models.PositiveIntegerField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="processing_started_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="queued_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="transcribed_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name="voicetasksession",
+            name="transcription_duration_ms",
+            field=models.PositiveIntegerField(blank=True, null=True),
+        ),
+    ]

plane-src/apps/api/plane/db/models/voice_tasker.py

@@ -128,6 +128,17 @@ class VoiceTaskSession(BaseModel):
     intent = models.CharField(max_length=40, blank=True)
     parsed_json = models.JSONField(blank=True, default=dict)
     client_context = models.JSONField(blank=True, default=dict)
+    queued_at = models.DateTimeField(null=True, blank=True)
+    processing_started_at = models.DateTimeField(null=True, blank=True)
+    transcribed_at = models.DateTimeField(null=True, blank=True)
+    completed_at = models.DateTimeField(null=True, blank=True)
+    failed_at = models.DateTimeField(null=True, blank=True)
+    transcription_duration_ms = models.PositiveIntegerField(null=True, blank=True)
+    parsing_duration_ms = models.PositiveIntegerField(null=True, blank=True)
+    processing_duration_ms = models.PositiveIntegerField(null=True, blank=True)
+    parser_prompt_tokens = models.PositiveIntegerField(null=True, blank=True)
+    parser_completion_tokens = models.PositiveIntegerField(null=True, blank=True)
+    parser_total_tokens = models.PositiveIntegerField(null=True, blank=True)
     created_task = models.ForeignKey(
         "db.Issue",
         on_delete=models.SET_NULL,