From c6ea0fe060401f688a00f93f7b14a2ad14d1a64a Mon Sep 17 00:00:00 2001 From: witbrock Date: Fri, 11 Sep 2026 00:17:28 -0400 Subject: [PATCH] feat(speech): add contextual live voice, client diagnostics and queue recovery --- docs/engineering/speech_io.md | 124 ++++++++- .../integrations/internal_mcp/orchestrator.py | 4 +- src/backend/languagemodels/model_defaults.py | 19 +- src/backend/server/routes/speech_routes.py | 136 +++++++++- src/backend/server/routes/von_routes.py | 64 ++++- src/backend/services/chat_history_service.py | 8 + .../services/chat_prompt_queue_service.py | 179 +++++++++++++ .../services/client_capabilities_service.py | 3 + .../services/model_registry_service.py | 4 +- .../services/speech_session_service.py | 162 ++++++++++++ .../services/speech_telemetry_service.py | 227 ++++++++++++++++ .../services/speech_transcription_service.py | 18 +- .../durable/model_selection_workflow.py | 4 +- .../web/von_interface/static/js/chatTab.js | 76 ++++-- .../static/js/clientCapabilitiesReporter.js | 2 + .../von_interface/static/js/clientContext.js | 67 +++++ .../web/von_interface/static/js/dictation.js | 68 ++++- .../static/js/streamingSpeech.js | 247 ++++++++++++++++++ .../static/js/voiceConversation.js | 102 ++++++++ .../web/von_interface/templates/chat_tab.html | 5 +- .../backend/test_chat_prompt_queue_routes.py | 33 +++ .../backend/test_chat_prompt_queue_service.py | 140 ++++++++++ tests/backend/test_speech_sessions.py | 211 +++++++++++++++ .../test_von_turn_execution_debug_info.py | 7 + .../test_workflow_model_selection_workflow.py | 4 +- tests/frontend/chatTaskQueue.test.js | 32 ++- tests/frontend/dictation.test.js | 3 +- tests/frontend/streamingSpeech.test.js | 95 +++++++ tests/frontend/voiceConversation.test.js | 46 ++++ 29 files changed, 2032 insertions(+), 58 deletions(-) create mode 100644 src/backend/services/speech_session_service.py create mode 100644 src/backend/services/speech_telemetry_service.py create mode 100644 src/frontend/web/von_interface/static/js/clientContext.js create mode 100644 src/frontend/web/von_interface/static/js/streamingSpeech.js create mode 100644 src/frontend/web/von_interface/static/js/voiceConversation.js create mode 100644 tests/backend/test_speech_sessions.py create mode 100644 tests/frontend/streamingSpeech.test.js create mode 100644 tests/frontend/voiceConversation.test.js diff --git a/docs/engineering/speech_io.md b/docs/engineering/speech_io.md index f6fa68495..66b6317ff 100644 --- a/docs/engineering/speech_io.md +++ b/docs/engineering/speech_io.md @@ -1,14 +1,18 @@ -# Recorded dictation and speech playback +# Contextual dictation and conversational speech - **Kind:** Bounded implementation and operating contract -- **Reviewed:** 10 September 2026 +- **Reviewed:** 11 September 2026 - **Owner / review trigger:** Speech workstream; review on audio-provider, authentication, browser capture or presenter changes. - **Current decisions:** [JVNAUTOSCI-888](https://naoinstitute.atlassian.net/browse/JVNAUTOSCI-888) owns the recorded-dictation repair; + native Von task `#V#task_deliver_contextual_streaming_dictation_and_709eb2df` + owns the current streaming, conversational speech and client diagnostics + delivery. Find it in Tasks by its title, **Deliver contextual streaming + dictation and conversational speech on mobile and desktop**. [JVNAUTOSCI-812](https://naoinstitute.atlassian.net/browse/JVNAUTOSCI-812) - owns the current conversational speech plan. Streaming voice is planned, - not provided by the block-dictation implementation. + remains historical planning context. The native programme was explicitly + requested; it does not change the project's migration writer. ## User flow @@ -20,6 +24,30 @@ releases the microphone and prevents late permission or transcription results from changing the draft. A failed upload retains audio in the active page and offers **Retry recording** without another microphone request. +When `gpt-live-transcribe` is enabled, live transcription becomes the initial +dictation choice. Provisional captions update as audio arrives. Finish commits +the outstanding buffer and waits for final text. Completed items are ordered by +their commit events and deduplicated by provider item ID. A stream failure adds +already completed text to the draft; an unfinished utterance may need repeating. +**More actions → Dictation** retains recorded audio and browser recognition as +explicit alternatives. Live audio has no retained recording to retry. + +Select **Start voice** (or **Alt+Shift+V**) for conversational mode. It opens a +conversation if needed, keeps listening, and submits completed utterances through +the existing durable prompt queue with stable submission IDs. Von's ordinary +agent, tools, scoped model choice and presenter channels produce each reply. +Its spoken channel is synthesised as streamed PCM. This starts audio before the +audio response finishes downloading; it does not start narration before the +ordinary agent has produced its reply. + +Speaking interrupts playback. Only the reply to the latest utterance in the +same active voice session may speak. **End voice** releases microphone and +playback immediately; changing conversation/organisation, hiding the page or +leaving it ends voice too. Already authorised tool work continues through its +normal queue and result path. Its Stop control remains available in the Thinking +card. Select Start voice to reconnect; reconnect never resubmits old utterances. +Typing and editable dictation remain available alongside voice. + Changing conversation or organisation cancels the capture. Leaving the page releases it; backgrounding finishes a recording. Raw audio is not saved by Von. Retry audio is lost when the page is closed or the capture is cancelled. @@ -40,6 +68,10 @@ in the scoped model pool. Add it as an additional model, not the primary conversational model. Audio models are excluded from automatic chat fallback and selection candidates. This does not change the main conversational model. No credentials or enabled-model settings are changed by this feature. +Live input additionally needs `gpt-live-transcribe`; conversational playback +needs `gpt-4o-mini-tts` (preferred) or `tts-1`. These are audio-only models and +are excluded from ordinary chat selection. Spoken replies are bounded to 4,096 +characters; an output failure preserves the full answer in chat. `GET /api/speech/capabilities` reports actor-specific availability. The composer refreshes it on initialisation, focus and preference changes. A configured key @@ -60,8 +92,9 @@ SDK retry. A failed request can be retried from retained browser audio. This is an initial operational bound, not a speech-turn duration or a latency claim; revisit it against actual successful recording/transcription durations. -Context is the recent visible conversation and current draft (up to 6,000 -characters), plus up to 80 visible literal concept names. The service does not +Context is the recent visible conversation, current draft and conversation focus +(up to 6,000 characters), plus up to 80 visible literal concept/focal task names. +Voice refreshes this context after replies. The service does not scan Vontology or fetch private conversations. Context/keywords are recognition hints, not required output or tool instructions. Transcription uses one audio model call and no semantic rewriting pass. `gpt-transcribe` receives plural @@ -82,6 +115,68 @@ Android or iOS/iPadOS. Mobile playback is invoked synchronously with the user gesture, and playback errors offer a visible retry. Automatic narration can still be blocked by device autoplay policy; tap **Speak** in that case. +Conversational voice uses Web Audio PCM playback, with the audio context opened +in the Start voice gesture. Playback queues at most two seconds ahead and aborts +both the fetch and scheduled audio nodes on interruption. It discloses the +provider and that Von's voice is AI-generated. + +## Streaming transport and endpointing + +`POST /api/speech/connection` authenticates the existing session/window actor, +checks the exact enabled audio model and exchanges a bounded SDP offer for a +transcription-only WebRTC session. The standard provider key remains on the +server. Browser audio goes directly to the provider; the connection has no Von +agent or tool authority. `POST /api/speech/speak` uses the same actor binding and +streams signed 16-bit mono PCM at 24 kHz. Provider streams close on completion or +client disconnect. + +Live provider verification found that `gpt-live-transcribe` rejects server-side +turn detection, despite the generic transcription guide referring to it. The +session therefore uses `turn_detection: null`. Browser audio analysis detects a +sustained onset (120 ms) and commits after 900 ms of quiet, using echo/noise +suppression and an adaptive energy floor. This is acoustic pause detection, +not a semantic judgement that the user has finished a thought. Brief gaps and +isolated clicks are covered by tests; natural pauses, soft speech, noisy rooms +and accents need physical-device evaluation before making quality claims. +Explicit Finish and End remain available. The initial transport bounds are 30 +seconds to establish WebRTC and 60 seconds for an unacknowledged final commit; +the latter releases a stuck paid media connection and preserves completed text. +Revisit these bounds using observed successful durations, not whole-task budgets. + +## Client and input diagnostics + +`client_context.v1` travels with each direct or queued prompt into actor-scoped +history debug and turn-execution diagnostics. It records a per-page client ID, +browser family/major, browser-exposed Client Hints, platform/version/model when +available, touch points, viewport, orientation, display mode, visibility, locale, +secure-context state and the versioned frontend asset path. Missing hardware +model information remains unknown; a reduced Android user agent is not proof of +a Pixel model. Client-supplied fields are bounded observations, never authority. + +Input events also survive failures before a chat message exists. +`POST /api/speech/attempts/` stores lifecycle, revision counts, +commit/submission IDs, engine/model/context sizes, audio format/rate and elapsed +times in `speech_input_attempts`. Exact actor-and-organisation GET read-back is +available at the same path. Events retain a maximum of 100 entries per attempt +and expire after 30 days; provisional browser revisions are throttled. The +snapshot includes backend version. Raw user-agent strings, transcripts and +audio are excluded from diagnostic storage. Full diagnostic histories keep their +existing actor-scoped visibility; shared transcript broadcasts omit them. + +## Interrupted queue recovery + +Legacy interrupted rows can be resumed through the existing requeue route. The +server first reconciles an exact prior assistant result. Otherwise it preserves +the original attempt and bounded tool/effect observations in the continuation, +retires the legacy row, and activates one idempotent server-owned successor +through the existing handoff service. Repeated resume requests return that same +successor. Unknown effect outcomes stay unknown; they are not evidence of failure +or permission to repeat a write. This repairs missing execution ownership, not +a general guarantee that every interrupted external effect is automatically +reconciled. Legacy client-owned queued rows say **Ready to resume**; **Next up** +is reserved for server-owned queue work. Queue rows and native Tasks remain +distinct objects. + ## Evidence boundary Baseline `59cbeb2f` had browser-only dictation, no vocabulary context, @@ -119,3 +214,20 @@ establish physical-device microphone or autoplay behaviour. Provider reference: [OpenAI file transcription](https://developers.openai.com/api/docs/guides/speech-to-text). Browser reference: [MDN SpeechRecognition](https://developer.mozilla.org/en-US/docs/Web/API/SpeechRecognition). + +On 11 September 2026, the new streaming path was exercised in installed Chrome +152 at 1280 × 900 and 390 × 844, using the actual composer template, browser +controllers and authenticated speech routes with a loopback synthetic actor/model +pool, synthetic WAV microphone and real OpenAI media providers. Both live +dictation and voice recognised “Von uses Vontology and Wikidata for research.” +The mobile-size voice replay submitted it once, streamed a fixture-supplied +reply, and End released every microphone track. Neither viewport overflowed. +A separate streamed synthesis check received its first 8,192 PCM bytes in +1,111 ms. These are bounded protocol, layout and lifecycle observations; they +do not measure ordinary-agent response latency, physical Pixel/iOS acoustics, +or deployment on the DGX. Targeted tests additionally exercise duplicate and +out-of-order finals, permission/cancel races, barge-in/late replies, queue +handoff/reconciliation, actor denial and telemetry isolation/read-back. + +Streaming references: [OpenAI realtime transcription](https://developers.openai.com/api/docs/guides/realtime-transcription), +[WebRTC transport](https://developers.openai.com/api/docs/guides/voice-webrtc?api=realtime). diff --git a/src/backend/integrations/internal_mcp/orchestrator.py b/src/backend/integrations/internal_mcp/orchestrator.py index 78c720599..a3f06bd9e 100644 --- a/src/backend/integrations/internal_mcp/orchestrator.py +++ b/src/backend/integrations/internal_mcp/orchestrator.py @@ -14310,9 +14310,9 @@ def _stage_model_candidates( str(entry.get("model") or "").strip() or None ) host = str(entry.get("host") or "").strip() or None - from ...languagemodels.model_defaults import is_transcription_model + from ...languagemodels.model_defaults import is_audio_only_model - if not provider or not model or is_transcription_model(provider, model): + if not provider or not model or is_audio_only_model(provider, model): continue model_parameters = ( normalise_model_parameters_for_storage( diff --git a/src/backend/languagemodels/model_defaults.py b/src/backend/languagemodels/model_defaults.py index 0ac7e3470..de81cd55d 100644 --- a/src/backend/languagemodels/model_defaults.py +++ b/src/backend/languagemodels/model_defaults.py @@ -13,10 +13,17 @@ TRANSCRIPTION_MODELS = ("gpt-transcribe", "gpt-4o-transcribe", "gpt-4o-mini-transcribe") -def is_transcription_model(provider: object, model: object) -> bool: - return str(provider or "").strip().lower() == "openai" and str( - model or "" - ).strip().removeprefix("openai:") in TRANSCRIPTION_MODELS +LIVE_TRANSCRIPTION_MODELS = ("gpt-live-transcribe",) +SPEECH_OUTPUT_MODELS = ("gpt-4o-mini-tts", "tts-1") + + +def is_audio_only_model(provider: object, model: object) -> bool: + return ( + str(provider or "").strip().lower() == "openai" + and str(model or "").strip().removeprefix("openai:") + in TRANSCRIPTION_MODELS + LIVE_TRANSCRIPTION_MODELS + SPEECH_OUTPUT_MODELS + ) + # --------------------------------------------------------------------------- # Ollama @@ -43,3 +50,7 @@ def is_transcription_model(provider: object, model: object) -> bool: META_MUSE_TEMPERATURE: float = 1.0 META_MUSE_TOP_P: float = 1.0 META_MUSE_MAX_OUTPUT_TOKENS: int = 32000 + + +# Compatibility for callers using the original audio exclusion helper. +is_transcription_model = is_audio_only_model diff --git a/src/backend/server/routes/speech_routes.py b/src/backend/server/routes/speech_routes.py index c1dbff429..57a9167f2 100644 --- a/src/backend/server/routes/speech_routes.py +++ b/src/backend/server/routes/speech_routes.py @@ -4,7 +4,15 @@ from datetime import datetime, timezone from typing import Any, Dict, Optional -from flask import Blueprint, current_app, jsonify, request, session +from flask import ( + Blueprint, + Response, + current_app, + jsonify, + request, + session, + stream_with_context, +) from ...services.chat_history_service import ( ChatHistoryServiceError, @@ -49,7 +57,13 @@ def speech_capabilities(): "available": False, "reason": "Speech configuration is temporarily unavailable.", } - response = jsonify(transcription=data) + from ...services.speech_session_service import media_capabilities + + try: + media = media_capabilities(actor, organisation) + except Exception: + media = {} + response = jsonify(transcription=data, **media) response.headers["Cache-Control"] = "private, no-store" return response @@ -282,3 +296,121 @@ def record_speech_telemetry(): "history_update": {"updated": updated, "reason": update_reason}, } ) + + +@speech_bp.route("/connection", methods=["POST"]) +def create_speech_connection(): + from ...services.speech_session_service import create_transcription_connection + + return _media_request( + create_transcription_connection, ("sdp", "context", "vocabulary", "language") + ) + + +def _media_request(operation, fields): + from ...languagemodels.llm_interface import ModelExecutionEligibilityError + from ...services.speech_transcription_service import SpeechUnavailable + + actor, organisation = _speech_actor() + if not actor: + return jsonify(error="authentication_required"), 401 + request.max_content_length = 100000 + payload = request.get_json(silent=True) + if not isinstance(payload, dict): + return jsonify(error="invalid_media_request"), 400 + try: + result = operation( + actor=actor, + organisation=organisation, + **{key: payload.get(key, "") for key in fields}, + ) + response = jsonify(success=True, **result) + response.headers["Cache-Control"] = "private, no-store" + return response + except ValueError as exc: + return jsonify(error="invalid_media_request", message=str(exc)), 400 + except (SpeechUnavailable, ModelExecutionEligibilityError) as exc: + return jsonify(error="speech_unavailable", message=str(exc)), 503 + except Exception as exc: + current_app.logger.warning("Speech connection failed (%s)", type(exc).__name__) + return jsonify( + error="speech_connection_failed", + message="Live speech could not connect. Try recorded dictation.", + ), 502 + + +@speech_bp.route("/speak", methods=["POST"]) +def stream_spoken_audio(): + from ...languagemodels.llm_interface import ModelExecutionEligibilityError + from ...services.speech_session_service import open_spoken_audio + from ...services.speech_transcription_service import SpeechUnavailable + + actor, organisation = _speech_actor() + if not actor: + return jsonify(error="authentication_required"), 401 + request.max_content_length = 50000 + payload = request.get_json(silent=True) or {} + try: + chunks, model, close = open_spoken_audio( + actor=actor, + organisation=organisation, + text=payload.get("text") if isinstance(payload, dict) else None, + ) + except ValueError as exc: + return jsonify(error="invalid_speech", message=str(exc)), 400 + except (SpeechUnavailable, ModelExecutionEligibilityError) as exc: + return jsonify(error="speech_unavailable", message=str(exc)), 503 + except Exception as exc: + current_app.logger.warning("Speech playback failed (%s)", type(exc).__name__) + return jsonify( + error="speech_failed", + message="Speech could not start. The response remains in chat.", + ), 502 + response = Response( + stream_with_context(chunks), mimetype="application/octet-stream" + ) + response.headers.update( + { + "Cache-Control": "private, no-store", + "X-Accel-Buffering": "no", + "X-Speech-Format": "pcm_s16le_24000", + "X-Speech-Model": model, + } + ) + response.call_on_close(close) + return response + + +@speech_bp.route("/attempts/", methods=["POST", "GET"]) +def speech_attempt(attempt_id): + from ...services.speech_telemetry_service import get_attempt, record_attempt + + actor, organisation = _speech_actor() + if not actor: + return jsonify(error="authentication_required"), 401 + request.max_content_length = 12000 + try: + if request.method == "POST": + result = record_attempt( + actor=actor, + organisation=organisation, + attempt_id=attempt_id, + payload=request.get_json(silent=True), + user_agent=request.headers.get("User-Agent"), + ) + else: + result = get_attempt( + actor=actor, organisation=organisation, attempt_id=attempt_id + ) + if result is None: + return jsonify(error="attempt_not_found"), 404 + except ValueError as exc: + return jsonify(error="invalid_attempt", message=str(exc)), 400 + except Exception as exc: + current_app.logger.warning( + "Speech diagnostics unavailable (%s)", type(exc).__name__ + ) + return jsonify(error="telemetry_unavailable", stored=False), 503 + response = jsonify(result) + response.headers["Cache-Control"] = "private, no-store" + return response diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index f25a85aca..c8bc53e56 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -1743,17 +1743,62 @@ def claim_chat_prompt_queue_route(queue_id: str): ) +def _queue_resume_conversation_binding(scope, session_id): + if not session_id: + raise chat_prompt_queue_service.InvalidChatPromptQueueInput( + "The interrupted prompt has no conversation to resume" + ) + actor = scope["user_concept_id"] + owner, invite = _resolve_shared_conversation_owner( + user_concept_id=actor, session_id=session_id + ) + owner = owner or actor + namespace = _canonical_conversation_history_namespace( + owner_user_id=owner, + actor_namespace=scope.get("namespace"), + shared_invite=invite, + ) + if chat_history_service.is_external_conversation_read_only( + user_id=owner, session_id=session_id, namespace=namespace + ): + raise chat_prompt_queue_service.InvalidChatPromptQueueInput( + "Continue the imported conversation before resuming work" + ) + return { + "history_user_id": owner, + "history_namespace": namespace, + "conversation_key": build_conversation_key( + owner_user_id=owner, + history_namespace=namespace, + conversation_session_id=session_id, + ), + } + + @von_bp.route("/api/chat_prompt_queue//requeue", methods=["POST"]) def requeue_chat_prompt_queue_route(queue_id: str): scope = _get_current_chat_prompt_queue_scope() if isinstance(scope, tuple): return scope try: - record = chat_prompt_queue_service.requeue_prompt_record( - scope=scope, - queue_id=queue_id, - current_server_instance_id=SERVER_INSTANCE_ID, - ) + payload = _chat_prompt_queue_payload() + if isinstance(payload.get("execution_envelope"), Mapping): + record = chat_prompt_queue_service.resume_legacy_prompt_record( + scope=scope, + queue_id=queue_id, + current_server_instance_id=SERVER_INSTANCE_ID, + execution_envelope=payload["execution_envelope"], + resolve_conversation=lambda session_id: ( + _queue_resume_conversation_binding(scope, session_id) + ), + ) + wake_chat_prompt_queue_dispatcher() + else: + record = chat_prompt_queue_service.requeue_prompt_record( + scope=scope, + queue_id=queue_id, + current_server_instance_id=SERVER_INSTANCE_ID, + ) return jsonify({"success": True, "item": record}) except Exception as exc: return _chat_prompt_queue_error_response( @@ -8514,6 +8559,15 @@ def _finalise_llm_debug_info( if not isinstance(llm_debug_info, dict): return llm_debug_info + from ...services.speech_telemetry_service import request_client_context + + client_context = request_client_context() + if client_context: + llm_debug_info["client_context"] = client_context + diagnostics = llm_debug_info.setdefault("turn_execution_diagnostics", {}) + if isinstance(diagnostics, dict): + diagnostics["client_context"] = client_context + tool_invocations_payload = ( llm_debug_info.get("tool_invocations") if isinstance(llm_debug_info.get("tool_invocations"), list) diff --git a/src/backend/services/chat_history_service.py b/src/backend/services/chat_history_service.py index af7a9ecd8..73557b31b 100644 --- a/src/backend/services/chat_history_service.py +++ b/src/backend/services/chat_history_service.py @@ -3976,6 +3976,14 @@ def add_message_to_history( else None ) + # Keep diagnostics in the existing actor-scoped debug channel, not the + # shared transcript. The request/enqueued envelope freezes its own client. + from .speech_telemetry_service import request_client_context + client_context = request_client_context() + if client_context: + llm_debug_data = llm_debug_data if isinstance(llm_debug_data, dict) else {} + llm_debug_data["client_context"] = client_context + # Add timestamp to message (and llm_debug_data if present) message_with_timestamp = {**message, "timestamp": datetime.now(timezone.utc)} author_user_id = message.get("author_user_id") diff --git a/src/backend/services/chat_prompt_queue_service.py b/src/backend/services/chat_prompt_queue_service.py index 63d118f47..d6f358aa3 100644 --- a/src/backend/services/chat_prompt_queue_service.py +++ b/src/backend/services/chat_prompt_queue_service.py @@ -57,6 +57,7 @@ MAX_EXECUTION_ENVELOPE_CHARS = 100_000 EXECUTION_ENVELOPE_ALLOWED_FIELDS = frozenset( { + "client_context", "initiation_id", "language", "model", @@ -452,6 +453,9 @@ def _normalise_execution_envelope(value: Any) -> dict[str, Any]: decoded = json.loads(encoded) if not isinstance(decoded, dict): # pragma: no cover - guarded above raise InvalidChatPromptQueueInput("execution_envelope must be an object") + if "client_context" in decoded: + from .speech_telemetry_service import sanitise_client_context + decoded["client_context"] = sanitise_client_context(decoded["client_context"]) return decoded @@ -3644,6 +3648,11 @@ def requeue_prompt_record( "completed_at": None, "last_error": None, "source": "restart", + "interrupted_attempt": existing.get("interrupted_attempt") or { + "client_request_id": existing.get("client_request_id"), + "attempt_id": existing.get("attempt_id"), + "server_instance_id": existing.get("server_instance_id"), + }, }, "$unset": { "active_conversation_key": "", @@ -4137,3 +4146,173 @@ def cancel_prompt_record( fallback_message="cancellable prompt was not found", ) return record + + +def resume_legacy_prompt_record( + *, + scope, + queue_id, + current_server_instance_id, + execution_envelope, + resolve_conversation=None, +): + """Reconcile an interrupted legacy turn, then use the existing durable handoff. + + A continuation carries the original request and exact receipt observations; + it is not a claim that an interrupted external effect failed or is replayable. + """ + query = {**_compatible_scope_query(scope), "queue_id": queue_id} + coll = _collection() + source = coll.find_one(query) + if not source: + raise ChatPromptQueueRecordNotFound("Interrupted prompt was not found") + successor = coll.find_one( + { + **_compatible_scope_query(scope), + "handoff_source_queue_id": queue_id, + "dispatch_mode": DISPATCH_MODE_SERVER, + } + ) + if successor: + if successor.get("status") == STATUS_QUEUED and not successor.get( + "dispatch_ready" + ): + return complete_server_dispatch_handoff( + scope=scope, queue_id=successor["queue_id"] + ) + return serialise_queue_record(successor) + if source.get("dispatch_mode") == DISPATCH_MODE_SERVER: + raise ChatPromptQueueReconciliationRequired( + "This server-owned turn must reconcile its executing attempt" + ) + if source.get("status") not in ACTIVE_STATUSES: + return serialise_queue_record(source) + if source.get("status") == STATUS_QUEUED and source.get("source") != "restart": + raise InvalidChatPromptQueueInput( + "Only interrupted work can use this continuation path" + ) + binding = ( + resolve_conversation(source.get("session_id")) if resolve_conversation else {} + ) + conversation_key = binding.get("conversation_key") or source.get("conversation_key") + if not conversation_key: + raise InvalidChatPromptQueueInput( + "Resolve the interrupted conversation before resuming work" + ) + attempt = source.get("interrupted_attempt") or { + key: source.get(key) + for key in ("client_request_id", "attempt_id", "server_instance_id") + } + request_id = attempt.get("client_request_id") + observed = None + if request_id: + from .chat_history_service import ( + build_chat_history_query, + get_chat_history_collection_service, + ) + + history = get_chat_history_collection_service(read_only=True) + if history is None: + raise ChatPromptQueueUnavailable("History reconciliation is unavailable") + completed = history.find_one( + { + **build_chat_history_query( + user_id=binding.get("history_user_id") or scope["user_concept_id"], + session_id=source.get("session_id"), + namespace=binding.get("history_namespace") + or scope.get("namespace"), + include_legacy=False, + ), + "history": { + "$elemMatch": { + "role": "assistant", + "llm_debug_data.request_id": request_id, + } + }, + }, + {"_id": 1}, + ) + if completed: + record = finish_prompt_record( + scope=scope, + queue_id=queue_id, + status=STATUS_COMPLETED, + attempt_id=source.get("attempt_id"), + ) + return {**record, "recovered_terminal": True} + from .turn_execution_record_service import get_turn_execution_record_projection + + record = get_turn_execution_record_projection( + request_id=request_id, namespace=scope.get("namespace") + ) + if ( + isinstance(record, Mapping) + and record.get("user_id") == scope["user_concept_id"] + ): + observed = { + key: record.get(key) + for key in ( + "request_id", + "terminal_status", + "tool_invocations", + "tool_observation_ledger", + "evidence_index", + "reference_manifest", + "required_effects", + "terminal_outcome_receipt", + "completion_report", + "effect_observation_journal", + ) + if record.get(key) is not None + } + # Preserve the exact locator even if a large record needs later reads. + if len(json.dumps(observed, default=str)) > 24000: + observed = { + "request_id": request_id, + "details": "available_through_exact_turn_telemetry", + } + if source.get("status") == STATUS_IN_PROGRESS: + requeue_prompt_record( + scope=scope, + queue_id=queue_id, + current_server_instance_id=current_server_instance_id, + ) + source = coll.find_one(query) + if source.get("conversation_key") != conversation_key: + # Reuse the canonical queued-record mutation before the handoff compares + # source and successor identity. Old restart rows can lack this binding. + update_queued_record( + scope=scope, + queue_id=queue_id, + session_id=source.get("session_id"), + conversation_key=conversation_key, + ) + source = coll.find_one(query) + envelope = _normalise_execution_envelope(execution_envelope) + envelope["workflow_inputs"] = { + **(envelope.get("workflow_inputs") or {}), + "interrupted_turn": { + "intent": "continue_from_observed_state", + "source_queue_id": queue_id, + "original_attempt": attempt, + "receipt_observations": observed, + "unobserved_effect_outcome": "unknown_not_failed", + "next_step": "reconcile_previous_outcomes_then_complete_remaining_work", + }, + } + replacement = create_queue_record( + scope=scope, + prompt_raw=source["prompt_raw"], + session_id=source.get("session_id"), + session_name=source.get("session_name"), + conversation_key=conversation_key, + source="interrupted_continuation", + dispatch_mode=DISPATCH_MODE_SERVER, + execution_envelope_version=1, + execution_envelope=envelope, + enqueue_submission_id="resume:" + queue_id, + handoff_source_queue_id=queue_id, + ) + return complete_server_dispatch_handoff( + scope=scope, queue_id=replacement["queue_id"] + ) diff --git a/src/backend/services/client_capabilities_service.py b/src/backend/services/client_capabilities_service.py index 6c0ded6f7..97e804ac2 100644 --- a/src/backend/services/client_capabilities_service.py +++ b/src/backend/services/client_capabilities_service.py @@ -102,6 +102,9 @@ def sanitise_client_capabilities( "user_agent_summary": _summarise_user_agent(user_agent), } + from .speech_telemetry_service import sanitise_client_context + snapshot["client_context"] = sanitise_client_context(payload.get("client_context"), user_agent) + speech_raw = payload.get("speech_synthesis") speech: dict[str, Any] = speech_raw if isinstance(speech_raw, dict) else {} diff --git a/src/backend/services/model_registry_service.py b/src/backend/services/model_registry_service.py index 9bf9f6331..4738ab47c 100644 --- a/src/backend/services/model_registry_service.py +++ b/src/backend/services/model_registry_service.py @@ -14,7 +14,7 @@ from typing import Any, Mapping, Optional, Sequence from urllib.parse import parse_qsl, unquote, urlsplit -from ..languagemodels.model_defaults import is_transcription_model +from ..languagemodels.model_defaults import is_audio_only_model from .settings_service import resolve_enabled_llm_settings, resolve_llm_setting logger = logging.getLogger(__name__) @@ -1299,7 +1299,7 @@ def _build_registry_from_settings() -> Mapping[str, Any]: continue provider = entry.get("provider") model = entry.get("model") - if is_transcription_model(provider, model): + if is_audio_only_model(provider, model): continue locality = "external" if isinstance(provider, str) and provider.lower() == "ollama": diff --git a/src/backend/services/speech_session_service.py b/src/backend/services/speech_session_service.py new file mode 100644 index 000000000..f75725add --- /dev/null +++ b/src/backend/services/speech_session_service.py @@ -0,0 +1,162 @@ +"""Actor-bound streaming media transport; no agent or tool authority lives here.""" + +from __future__ import annotations + +import json +import os + +import httpx + +from ..languagemodels.llm_interface import assert_model_execution_allowed +from ..languagemodels.model_defaults import ( + LIVE_TRANSCRIPTION_MODELS, + SPEECH_OUTPUT_MODELS, +) +from .settings_service import get_openai_env_var, resolve_enabled_llm_settings +from .speech_transcription_service import SpeechUnavailable, normalise_context + + +def media_model(actor, organisation, candidates): + enabled = resolve_enabled_llm_settings( + user_concept_id=actor, org_concept_id=organisation + ) + names = { + str(e.get("model", "")).removeprefix("openai:") + for e in enabled + if isinstance(e, dict) and e.get("provider") == "openai" + } + model = next((name for name in candidates if name in names), None) + if not model: + raise SpeechUnavailable( + "Enable a speech model in Settings: " + ", ".join(candidates) + "." + ) + if not os.environ.get(get_openai_env_var() or "OPENAI_API_KEY"): + raise SpeechUnavailable("The server's configured OpenAI key is unavailable.") + return model + + +def media_capabilities(actor, organisation): + result = {} + for kind, candidates in ( + ("streaming", LIVE_TRANSCRIPTION_MODELS), + ("speech_output", SPEECH_OUTPUT_MODELS), + ): + try: + result[kind] = { + "available": True, + "model": media_model(actor, organisation, candidates), + "provider": "openai", + } + except SpeechUnavailable as exc: + result[kind] = {"available": False, "reason": str(exc)} + return result + + +def _authorise(actor, organisation, candidates): + model = media_model(actor, organisation, candidates) + assert_model_execution_allowed( + provider="openai", + model=model, + user_concept_id=actor, + org_concept_id=organisation, + allow_ambient_actor_scope=False, + ) + return model + + +def create_transcription_connection( + *, actor, organisation, sdp, context="", vocabulary=None, language="" +): + # The server selects session type and model; the client cannot mint a general + # realtime agent credential or choose tools/instructions for an audio model. + if not isinstance(sdp, str) or not sdp.startswith("v=0") or len(sdp) > 65536: + raise ValueError("A bounded WebRTC offer is required.") + model = _authorise(actor, organisation, LIVE_TRANSCRIPTION_MODELS) + context, terms = normalise_context(context, vocabulary) + from .speech_transcription_service import normalise_language + + language = normalise_language(language) + transcription = { + "model": model, + "prompt": context, + "keywords": terms, + "delay": "low", + } + if language: + transcription["languages"] = [language] + config = { + "type": "transcription", + "audio": { + "input": { + "transcription": transcription, + "noise_reduction": {"type": "near_field"}, + # gpt-live-transcribe rejects server-side turn detection. + # The browser commits pauses and explicit Finish actions. + "turn_detection": None, + } + }, + } + with httpx.Client(timeout=30, follow_redirects=False) as client: + response = client.post( + "https://api.openai.com/v1/realtime/calls", + headers={ + "Authorization": "Bearer " + + os.environ[get_openai_env_var() or "OPENAI_API_KEY"] + }, + files={"sdp": (None, sdp), "session": (None, json.dumps(config))}, + ) + response.raise_for_status() + answer = response.text + if not answer.startswith("v=0") or len(answer) > 65536: + raise ValueError("The speech provider returned an invalid connection answer.") + return { + "sdp": answer, + "model": model, + "provider": "openai", + "vocabulary_count": len(terms), + "context_chars": len(context), + } + + +def open_spoken_audio(*, actor, organisation, text): + """Open before Flask sends headers, then close on completion/disconnect.""" + import openai + + if not isinstance(text, str) or not text.strip() or len(text) > 4096: + raise ValueError("Spoken text must contain 1–4096 characters.") + model = _authorise(actor, organisation, SPEECH_OUTPUT_MODELS) + client = openai.OpenAI( + api_key=os.environ[get_openai_env_var() or "OPENAI_API_KEY"], + timeout=120, + max_retries=0, + ) + manager = client.audio.speech.with_streaming_response.create( + model=model, + voice="alloy" if model == "tts-1" else "coral", + input=text, + response_format="pcm", + ) + try: + response = manager.__enter__() + except BaseException: + client.close() + raise + + closed = False + + def close(): + nonlocal closed + if not closed: + closed = True + try: + manager.__exit__(None, None, None) + finally: + client.close() + + def chunks(): + try: + yield from response.iter_bytes(chunk_size=8192) + finally: + close() + + return chunks(), model, close diff --git a/src/backend/services/speech_telemetry_service.py b/src/backend/services/speech_telemetry_service.py new file mode 100644 index 000000000..9876957d9 --- /dev/null +++ b/src/backend/services/speech_telemetry_service.py @@ -0,0 +1,227 @@ +"""Bounded actor-scoped input diagnostics. No audio or transcript contents.""" + +from __future__ import annotations + +import hashlib +import math +import re +import threading +from datetime import UTC, datetime, timedelta + +from ..db.mongo_client import get_db + +_indexed = set() +_index_lock = threading.Lock() + + +def sanitise_client_context(value, user_agent=None): + from .client_capabilities_service import _summarise_user_agent + + value = value if isinstance(value, dict) else {} + out = { + "schema_version": "client_context.v1", + "client_reported": True, + "user_agent_summary": _summarise_user_agent( + user_agent + or ( + value.get("user_agent") + if isinstance(value.get("user_agent"), str) + else None + ) + ), + } + original_summary = value.get("user_agent_summary") + if out["user_agent_summary"]["browser_family"] == "Other" and isinstance( + original_summary, dict + ): + family = original_summary.get("browser_family") + major = original_summary.get("major_version") + if family in {"Chrome", "Safari", "Edge", "Firefox", "Other"}: + out["user_agent_summary"] = { + "browser_family": family, + "major_version": major + if isinstance(major, int) and 0 < major < 10000 + else None, + } + for key in ( + "client_id", + "captured_at", + "platform", + "platform_version", + "device_model", + "orientation", + "display_mode", + "visibility", + "frontend_asset", + "language", + "speech_item_id", + ): + raw = value.get(key) + if isinstance(raw, str): + out[key] = raw[:160] + for key in ("touch_points", "viewport_width", "viewport_height"): + raw = value.get(key) + if ( + isinstance(raw, (int, float)) + and not isinstance(raw, bool) + and math.isfinite(raw) + ): + out[key] = max(0, min(20000, int(raw))) + for key in ("mobile", "secure_context"): + if isinstance(value.get(key), bool): + out[key] = value[key] + out["brands"] = [ + {"brand": b["brand"][:80], "version": b["version"][:40]} + for b in (value.get("brands") if isinstance(value.get("brands"), list) else [])[ + :5 + ] + if isinstance(b, dict) + and isinstance(b.get("brand"), str) + and isinstance(b.get("version"), str) + ] + out["speech_attempt_ids"] = ( + [ + v[:100] + for v in (value.get("speech_attempt_ids") or [])[:8] + if isinstance(v, str) + ] + if isinstance(value.get("speech_attempt_ids"), list) + else [] + ) + return out + + +def request_client_context(): + from flask import has_request_context, request + + if not has_request_context(): + return None + payload = request.get_json(silent=True) if request.is_json else None + raw = payload.get("client_context") if isinstance(payload, dict) else None + if not isinstance(raw, dict): + return None + return sanitise_client_context(raw, request.headers.get("User-Agent")) + + +def _collection(): + db = get_db() + if db is None: + raise RuntimeError("Speech telemetry storage is unavailable") + coll = db["speech_input_attempts"] + key = (id(db.client), db.name) + with _index_lock: + if key not in _indexed: + coll.create_index("expires_at", expireAfterSeconds=0) + _indexed.add(key) + return coll + + +def _key(actor, organisation, attempt_id): + if not isinstance(attempt_id, str) or not re.fullmatch( + r"[A-Za-z0-9_-]{8,100}", attempt_id + ): + raise ValueError("Invalid speech attempt ID") + return hashlib.sha256( + (actor + "|" + (organisation or "") + "|" + attempt_id).encode() + ).hexdigest() + + +def record_attempt(*, actor, organisation, attempt_id, payload, user_agent=None): + key = _key(actor, organisation, attempt_id) + payload = payload if isinstance(payload, dict) else {} + raw = payload.get("event") + raw = raw if isinstance(raw, dict) else {} + allowed = { + "state", + "started", + "ended", + "error", + "revision", + "committed", + "submitted", + "speech_started", + "speech_stopped", + "playback_started", + "playback_ended", + "interrupted", + "recovery", + } + if raw.get("name") not in allowed: + raise ValueError("Unknown speech event") + event = {"name": raw["name"]} + for name in ( + "state", + "engine", + "model", + "reason", + "item_id", + "request_id", + "submission_id", + "format", + "context_version", + ): + if isinstance(raw.get(name), str): + event[name] = raw[name][:120] + for name in ( + "sequence", + "elapsed_ms", + "result_index", + "result_count", + "text_length", + "audio_bytes", + "vocabulary_count", + "context_chars", + "sample_rate", + "channel_count", + ): + value = raw.get(name) + if ( + isinstance(value, (int, float)) + and not isinstance(value, bool) + and math.isfinite(value) + ): + event[name] = max(0, min(100000000, int(value))) + if isinstance(raw.get("final"), bool): + event["final"] = raw["final"] + from .runtime_code_version_service import get_runtime_code_version + + now = datetime.now(UTC) + event["received_at"] = now + conversation_id = payload.get("conversation_id") + coll = _collection() + coll.update_one( + {"_id": key}, + { + "$setOnInsert": { + "actor": actor, + "organisation": organisation, + "attempt_id": attempt_id, + "client_context": sanitise_client_context( + payload.get("client_context"), user_agent + ), + "conversation_id": conversation_id[:100] + if isinstance(conversation_id, str) + else None, + "backend_version": get_runtime_code_version(), + "created_at": now, + }, + "$set": {"updated_at": now, "expires_at": now + timedelta(days=30)}, + "$push": {"events": {"$each": [event], "$slice": -100}}, + }, + upsert=True, + ) + return { + "stored": True, + "attempt_id": attempt_id, + "event_sequence": event.get("sequence"), + } + + +def get_attempt(*, actor, organisation, attempt_id): + return _collection().find_one( + { + "_id": _key(actor, organisation, attempt_id), + "expires_at": {"$gt": datetime.now(UTC)}, + }, + {"_id": 0}, + ) diff --git a/src/backend/services/speech_transcription_service.py b/src/backend/services/speech_transcription_service.py index c2f052767..c1d86f589 100644 --- a/src/backend/services/speech_transcription_service.py +++ b/src/backend/services/speech_transcription_service.py @@ -74,6 +74,15 @@ def normalise_context(context, vocabulary) -> tuple[str, list[str]]: return context, terms +def normalise_language(language): + language = language.strip() if isinstance(language, str) else "" + if language and not re.fullmatch(r"[a-zA-Z]{2,3}(?:-[a-zA-Z0-9]{2,8})*", language): + raise ValueError( + "Use a language tag such as en-NZ, mi or zh-CN, or leave it empty." + ) + return language.split("-", 1)[0].lower() + + def transcribe_audio( *, actor, organisation, audio, mime_type, context="", vocabulary=None, language="" ): @@ -104,19 +113,16 @@ def transcribe_audio( "prompt": prompt, } language = language.strip() if isinstance(language, str) else "" - if language and not re.fullmatch(r"[a-zA-Z]{2,3}(?:-[a-zA-Z0-9]{2,8})*", language): - raise ValueError( - "Use a language tag such as en-NZ, mi or zh-CN, or leave it empty." - ) + provider_language = normalise_language(language) if model == "gpt-transcribe": options["extra_body"] = { "keywords": terms, # The provider accepts language codes, not browser locales: a live # en-NZ request is rejected while the otherwise identical en works. - **({"languages": [language.split("-", 1)[0].lower()]} if language else {}), + **({"languages": [provider_language]} if provider_language else {}), } elif language: - options["language"] = language.split("-", 1)[0].lower() + options["language"] = provider_language start = time.monotonic() # Explicit transport bound protects occupied HTTP workers; no client timer # discards a usable result. Users can cancel or retry the retained recording. diff --git a/src/backend/workflows/durable/model_selection_workflow.py b/src/backend/workflows/durable/model_selection_workflow.py index 4e19c649e..aa7dd9a32 100644 --- a/src/backend/workflows/durable/model_selection_workflow.py +++ b/src/backend/workflows/durable/model_selection_workflow.py @@ -129,9 +129,9 @@ def _enabled_model_pool( provider = _context_string(entry.get("provider")) model = _context_string(entry.get("model")) host = _context_string(entry.get("host")) - from ...languagemodels.model_defaults import is_transcription_model + from ...languagemodels.model_defaults import is_audio_only_model - if not provider or not model or is_transcription_model(provider, model): + if not provider or not model or is_audio_only_model(provider, model): continue key = (provider.lower(), model, host) if key in seen: diff --git a/src/frontend/web/von_interface/static/js/chatTab.js b/src/frontend/web/von_interface/static/js/chatTab.js index 9fd4b108c..20a0e680d 100644 --- a/src/frontend/web/von_interface/static/js/chatTab.js +++ b/src/frontend/web/von_interface/static/js/chatTab.js @@ -1,3 +1,5 @@ +import { createVoiceConversation } from './voiceConversation.js'; +import { getClientContext } from './clientContext.js'; import { createDictationController } from './dictation.js'; import { renderImageAttachments, renderImageComposer, uploadConversationImage, imagesBlocked, takeImages, restoreImages, descriptorsForIds } from "./utils/conversationImages.js"; // Chat Tab Module @@ -21270,6 +21272,7 @@ function setPromptComposerValue(value, options = {}) { } promptInput.value = typeof value === 'string' ? value : ''; + if (!promptInput.value) dictationController?.clearAttemptIds?.(); dispatchPromptComposerInputEvent(promptInput); const hasSelection = @@ -24604,6 +24607,8 @@ const CHAT_STT_CONTINUOUS_STORAGE_KEY = 'chatSttContinuous'; const CHAT_STT_INTERIM_RESULTS_STORAGE_KEY = 'chatSttInterimResults'; let dictationController = null; +let voiceConversation = null; +const voicePromptQueueIds = new Set(); let dictationSendPending = false; let activeTtsTurnId = null; @@ -25038,6 +25043,7 @@ function toggleSpeakTurn(turnId, text, button) { } function stopDictation() { + voiceConversation?.end(); dictationController?.cancel(); } @@ -25046,14 +25052,18 @@ function getDictationContext() { const history = getConversationTranscriptTurnsSnapshot().slice(-4) .map(turn => turn.message || turn.text || '').join('\n').slice(-4500); const vocabulary = ['Von', 'Vontology']; + const focusNames = normaliseFocusedConcepts(activeChatSessionFocus).map(item => item.displayName); + vocabulary.push(...focusNames); document.querySelectorAll('#scrollableField [data-full-concept-id], .prompt-input-overlay [data-full-concept-id]') .forEach(element => { const name = element.querySelector('.vontology-cartouche-name')?.textContent || element.textContent; if (name?.trim()) vocabulary.push(name.trim()); }); return { + sessionId: activeChatSessionId, key: `${getCurrentUserConceptId() || ''}:${activeChatSessionId || ''}:${getSessionScopedNamespace() || ''}`, - text: `${history}\n${prompt}`.slice(-6000), vocabulary: [...new Set(vocabulary)].slice(0, 80), + text: `${focusNames.length ? 'Discussing: ' + focusNames.join(', ') + '\n' : ''}${history}\n${prompt}`.slice(-6000), + vocabulary: [...new Set(vocabulary)].slice(0, 80), ...getChatSpeechSettings().stt }; } @@ -28460,6 +28470,7 @@ async function promptRenameChatSession(sessionId, currentName) { } async function createChatSession(sessionName, options = {}) { + if (activeChatSessionId) stopDictation(); if (pendingChatOrganisationSwitchId !== null) { const error = new Error('Organisation change in progress.'); error.organisationSwitchSuperseded = true; @@ -34271,10 +34282,37 @@ export function initializeChatTab() { getContext: getDictationContext, fetchImpl: (url, options = {}) => fetch(url, { ...options, headers: buildChatFetchHeaders(options.headers) }), setValue: value => setPromptComposerValue(value, { promptInput }), - onStart: () => { stopSpeaking(); clearActiveTtsUi(); } + onStart: () => { voiceConversation?.end(); stopSpeaking(); clearActiveTtsUi(); } }); } + voiceConversation?.dispose(); + const voiceButton = document.getElementById('voiceConversationButton'); + if (voiceButton) voiceConversation = createVoiceConversation({ + button: voiceButton, status: document.getElementById('voiceConversationStatus'), + getContext: getDictationContext, + fetchImpl: (url, options = {}) => fetch(url, { ...options, headers: buildChatFetchHeaders(options.headers) }), + onStart: async () => { + dictationController?.cancel(); stopSpeaking(); clearActiveTtsUi(); + if (!activeChatSessionId && !await createNewChatSession()) { + throw new Error('A conversation could not be opened. Start voice to retry.'); + } + }, + onSubmit: async (text, speech) => { + const envelope = buildChatPromptExecutionEnvelope(); + envelope.client_context = { ...envelope.client_context, + speech_attempt_ids: [speech.attemptId], speech_item_id: speech.itemId }; + const row = await queuePromptForLater(text, { sessionId: activeChatSessionId, + sessionName: activeChatSessionName, enqueueSubmissionId: speech.submissionId, + executionEnvelope: envelope }); + if (!row?.queueId || row.enqueueFailed) throw new Error('Voice message awaiting queue recovery'); + voicePromptQueueIds.add(row.queueId); + if (voicePromptQueueIds.size > 256) voicePromptQueueIds.delete(voicePromptQueueIds.values().next().value); + syncChatPromptQueuePolling(); + syncServerDispatchedRetryObserver(); + } + }); + // Add event listeners sendButton.addEventListener('click', handleSendPrompt); resetButton.addEventListener('click', handleResetContext); @@ -35222,6 +35260,7 @@ function normaliseChatPromptQueueEntry(rawEntry, fallback = {}) { return { id, queueId, + source: String(rawEntry.source ?? fallback.source ?? ''), promptRaw: String(rawEntry.prompt_raw ?? rawEntry.promptRaw ?? fallback.promptRaw ?? ''), status: normaliseChatPromptQueueStatus(rawEntry.status ?? fallback.status), sessionId, @@ -35681,7 +35720,7 @@ function activeConversationQueueStateChanged(previousEntries, nextEntries) { function isObservableServerDispatchedRetry(entry) { return Boolean( entry?.queueId - && entry?.retrySourceQueueId + && (entry?.retrySourceQueueId || entry?.handoffSourceQueueId || voicePromptQueueIds.has(entry?.queueId)) && entry?.clientRequestId && entry?.attemptId && entry.status === CHAT_PROMPT_QUEUE_STATUS_IN_PROGRESS @@ -35857,6 +35896,7 @@ function buildChatPromptExecutionEnvelope({ resolveLocalRequestedLlm() ); return { + client_context: { ...getClientContext(), speech_attempt_ids: dictationController?.getAttemptIds?.() || [] }, ...(imageAttachmentIds.length ? { image_attachment_ids: imageAttachmentIds } : {}), ...(turnKind ? { turn_kind: turnKind } : {}), ...(initiationId ? { initiation_id: initiationId } : {}), @@ -36242,8 +36282,10 @@ async function requeuePersistedChatPromptQueueEntry(entry) { return entry; } const data = await fetchChatPromptQueueJson(`/${encodeURIComponent(entry.queueId)}/requeue`, { - method: 'POST' + method: 'POST', + body: JSON.stringify({ execution_envelope: buildChatPromptExecutionEnvelope() }) }); + if (data.item?.recovered_terminal) refreshActiveConversationHistoryAfterQueueChange(); return data.item ? normaliseChatPromptQueueEntry(data.item, entry) : entry; } @@ -36452,11 +36494,10 @@ function ensureChatTaskQueuePanel() { void (async () => { try { const requeued = await requeuePersistedChatPromptQueueEntry(queued); - mergePersistedChatPromptQueueEntry({ - ...queued, - ...requeued, - status: CHAT_PROMPT_QUEUE_STATUS_QUEUED - }); + if (requeued.queueId !== queued.queueId) { + queuedChatPrompts = queuedChatPrompts.filter(entry => entry.queueId !== queued.queueId); + } + mergePersistedChatPromptQueueEntry({ ...queued, ...requeued }); renderChatTaskQueuePanel(); refreshChatSessionTabActivityIndicators(); updateSendButtonForCurrentChatState(); @@ -36762,10 +36803,10 @@ function renderChatTaskQueuePanel() { } else if (isFailed) { label.textContent = `Failed • ${sessionLabel}`; } else { - queuedOrdinal += 1; - label.textContent = queuedOrdinal === 1 - ? `Next up • ${sessionLabel}` - : `Queue #${queuedOrdinal} • ${sessionLabel}`; + if (isServerDispatched) queuedOrdinal += 1; + label.textContent = !isServerDispatched + ? `Ready to resume • ${sessionLabel}` + : queuedOrdinal === 1 ? `Next up • ${sessionLabel}` : `Queue #${queuedOrdinal} • ${sessionLabel}`; } const editor = document.createElement('textarea'); @@ -36807,7 +36848,7 @@ function renderChatTaskQueuePanel() { `Complete queue handoff ${index + 1}` ); actions.appendChild(retryHandoffButton); - } else if (isInterrupted) { + } else if (isInterrupted || (!isServerDispatched && entry.status === CHAT_PROMPT_QUEUE_STATUS_QUEUED && entry.source === 'restart')) { const restartButton = document.createElement('button'); restartButton.type = 'button'; restartButton.className = 'btn-mini chat-task-queue-restart'; @@ -37160,7 +37201,7 @@ async function handleSendPrompt(options = {}) { && !hasPromptOverride && !selectedQueueEntry ); - if (directComposerSubmission && dictationController?.getState() !== 'idle' && dictationController) { + if (directComposerSubmission && dictationController?.hasPendingInput()) { if (dictationSendPending) return; const scopeBefore = getDictationContext().key; dictationSendPending = true; @@ -38626,7 +38667,9 @@ function appendMessage(sender, message, turnId, hasLlmDebug = false, isHistory = }); // Auto-speak new assistant responses when enabled. - if (!isHistory && turnId && ttsSupported && isChatTtsEnabled()) { + if (!isHistory && turnId && voiceConversation?.isActive()) { + void voiceConversation.reply({ text: spokenTextForTurn, turnId, clientContext: debugData?.client_context }); + } else if (!isHistory && turnId && ttsSupported && isChatTtsEnabled()) { // Auto-speak uses the talk track only (never the screen channel). if (spokenTextForTurn && spokenTextForTurn.trim()) { toggleSpeakTurn(turnId, spokenTextForTurn, speakButton); @@ -41925,6 +41968,7 @@ export function __testOnly_resetChatRequestState() { queuedChatPromptClaimsInFlight.clear(); queuedChatPromptSubmissionsInFlight.clear(); serverDispatchedRetryObserversInFlight.clear(); + voicePromptQueueIds.clear(); chatPromptQueueAdmissionServerInstanceId = null; for (const timer of queuedChatPromptSyncTimers.values()) { clearTimeout(timer); diff --git a/src/frontend/web/von_interface/static/js/clientCapabilitiesReporter.js b/src/frontend/web/von_interface/static/js/clientCapabilitiesReporter.js index 6f6755321..55b0d0ca8 100644 --- a/src/frontend/web/von_interface/static/js/clientCapabilitiesReporter.js +++ b/src/frontend/web/von_interface/static/js/clientCapabilitiesReporter.js @@ -1,3 +1,4 @@ +import { getClientContext } from './clientContext.js'; // Client capability reporting (browser-only) // // JVNAUTOSCI-954: Safely report bounded, non-authoritative client capability hints @@ -106,6 +107,7 @@ export function __testOnly_buildClientCapabilitiesPayload() { return { client_reported_timestamp: safeIsoNow(), + client_context: getClientContext(), speech_synthesis: { supported: ttsSupported, voices_count: Array.isArray(voices) ? voices.length : 0, diff --git a/src/frontend/web/von_interface/static/js/clientContext.js b/src/frontend/web/von_interface/static/js/clientContext.js new file mode 100644 index 000000000..4c9c34f31 --- /dev/null +++ b/src/frontend/web/von_interface/static/js/clientContext.js @@ -0,0 +1,67 @@ +// Bounded diagnostics only. Client identity never supplies actor authority. +let hints = null; +let hintsRequest = null; +let clientId = null; +export function newSpeechId(root = globalThis) { + return root.crypto?.randomUUID?.() || 'speech-' + Date.now() + '-' + Math.random().toString(36).slice(2); +} +export function refreshClientHints(root = globalThis) { + if (!hintsRequest && root.navigator?.userAgentData?.getHighEntropyValues) { + hintsRequest = root.navigator.userAgentData.getHighEntropyValues(['model', 'platformVersion', 'fullVersionList']) + .then(value => { hints = value; }).catch(() => {}); + } + return hintsRequest || Promise.resolve(); +} +// Resolve optional hints while the page opens, before the first speech attempt. +void refreshClientHints(); +export function getClientContext(root = globalThis) { + if (!clientId) clientId = newSpeechId(root); + const nav = root.navigator || {}; + const ua = hints || nav.userAgentData || {}; + void refreshClientHints(root); + const script = root.document?.querySelector?.('script[type="module"][src*="main.js"]'); + return { + client_id: clientId, + user_agent: nav.userAgent || null, + captured_at: new Date().toISOString(), + brands: (ua.fullVersionList || ua.brands || []).slice(0, 5), + platform: ua.platform || nav.platform || null, + platform_version: ua.platformVersion || null, + device_model: ua.model || null, + mobile: typeof ua.mobile === 'boolean' ? ua.mobile : null, + touch_points: nav.maxTouchPoints || 0, + viewport_width: root.innerWidth || null, viewport_height: root.innerHeight || null, + orientation: root.screen?.orientation?.type || null, + display_mode: root.matchMedia?.('(display-mode: standalone)').matches ? 'standalone' : 'browser', + secure_context: root.isSecureContext === true, + visibility: root.document?.visibilityState || null, + frontend_asset: script?.getAttribute('src') || null, + language: nav.language || null + }; +} +export function createSpeechReporter({ fetchImpl = (...args) => fetch(...args), getContext = () => ({}), root = globalThis } = {}) { + const attemptId = newSpeechId(root); + const snapshot = getClientContext(root); + const started = root.performance?.now?.() ?? Date.now(); + let sequence = 0; + let lastState = null; + let lastRevisionAt = -Infinity; + return { + attemptId, + event(name, values = {}) { + const context = getContext(); + const now = root.performance?.now?.() ?? Date.now(); + if (name === 'revision' && !values.final && now - lastRevisionAt < 500) return; + if (name === 'revision') lastRevisionAt = now; + // Report structural lifecycle/revisions, never recognition text/audio. + const event = { name, sequence: sequence++, elapsed_ms: Math.round(now - started), ...values }; + if (name === 'state' && values.state === lastState) return; + if (name === 'state') lastState = values.state; + void fetchImpl('/api/speech/attempts/' + encodeURIComponent(attemptId), { + method: 'POST', credentials: 'same-origin', keepalive: true, + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ client_context: snapshot, conversation_id: context.sessionId || null, event }) + }).catch(() => {}); + } + }; +} diff --git a/src/frontend/web/von_interface/static/js/dictation.js b/src/frontend/web/von_interface/static/js/dictation.js index 4f7c9d473..7ba2f50f2 100644 --- a/src/frontend/web/von_interface/static/js/dictation.js +++ b/src/frontend/web/von_interface/static/js/dictation.js @@ -1,3 +1,5 @@ +import { createStreamingInput } from './streamingSpeech.js'; +import { createSpeechReporter } from './clientContext.js'; import { isSpeechRecognitionSupported, startSpeechRecognition, stopSpeaking } from './speech.js'; const RECORDING_TYPES = ['audio/webm;codecs=opus', 'audio/mp4', 'audio/webm']; @@ -38,9 +40,12 @@ export function createDictationController({ input, button, status, cancelButton, let capability = null; let state = 'idle'; let disposed = false; + const attemptIds = []; + let engineChosen = false; function render(next, message = '') { state = next; + current?.reporter?.event('state', { state: next }); const busy = ['requesting', 'recording', 'transcribing'].includes(next); button.textContent = next === 'recording' ? 'Finish dictation' : next === 'transcribing' ? 'Transcribing…' : next === 'requesting' ? 'Opening microphone…' : 'Dictate'; button.disabled = next === 'transcribing' || next === 'requesting'; @@ -65,6 +70,8 @@ export function createDictationController({ input, button, status, cancelButton, const capture = current; current = null; if (capture) { + capture.reporter?.event('ended', { reason: 'cancelled' }); + capture.live?.cancel(); capture.abort?.abort(); capture.recognition?.abort(); if (capture.recorder?.state === 'recording') capture.recorder.stop(); @@ -74,9 +81,14 @@ export function createDictationController({ input, button, status, cancelButton, render('idle'); } - function fail(capture, message) { + function fail(capture, message, reason = 'input_failed') { if (!valid(capture)) return; release(capture); + capture.reporter?.event('error', { reason }); + if (!capture.blob) { + current = null; + capture.recognition?.abort(); + } render('error', message); capture.resolve?.(false); } @@ -86,6 +98,11 @@ export function createDictationController({ input, button, status, cancelButton, const value = String(input.value || ''); const selection = value === capture.base ? capture.selection : null; setValue(insertDictation(value, text.trim(), selection)); + if (text.trim() && capture.reporter) { + attemptIds.push(capture.reporter.attemptId); + if (attemptIds.length > 8) attemptIds.shift(); + } + capture.reporter?.event('committed', { text_length: text.trim().length, final: true }); current = null; render('idle', text.trim() ? 'Dictation added. Edit it or send when ready.' : 'No speech was detected. Try again.'); capture.resolve?.(!!text.trim()); @@ -121,10 +138,28 @@ export function createDictationController({ input, button, status, cancelButton, selection: { start: input.selectionStart, end: input.selectionEnd }, chunks: [], bytes: 0 }; capture.completion = new Promise(resolve => { capture.resolve = resolve; }); current = capture; + capture.reporter = createSpeechReporter({ fetchImpl, getContext: () => capture.context, root }); + capture.reporter.event('started', { engine: engineSelect.value }); onStart(); render('requesting', 'Allow microphone access to dictate.'); try { - if (engineSelect.value === 'browser') { + if (engineSelect.value === 'streaming') { + if (!capability?.streaming?.available) throw new Error(capability?.streaming?.reason || 'Live transcription is unavailable. Choose recorded audio.'); + capture.live = createStreamingInput({ context, fetchImpl, root, + onPartial: text => { if (valid(capture)) status.textContent = text || 'Listening…'; }, + onFinal: () => { if (valid(capture)) status.textContent = capture.live.completedText(); }, + onState: (next, message) => { + if (!valid(capture)) return; + if (next === 'error') { + const recovered = capture.live.completedText(); + if (recovered) { commit(capture, recovered); status.textContent = message + ' Completed text added to your draft.'; } + else fail(capture, message); + } else render(next === 'listening' ? 'recording' : next === 'finishing' ? 'transcribing' : 'requesting', message); + }, + onEvent: (name, values) => capture.reporter.event(name, values) + }); + await capture.live.start(); + } else if (engineSelect.value === 'browser') { if (!isSpeechRecognitionSupported()) throw new Error('Browser recognition is unavailable. Choose recorded audio.'); capture.recognition = startSpeechRecognition({ language: context.language, continuous: context.continuous ?? false, interimResults: context.interimResults ?? true, @@ -132,9 +167,10 @@ export function createDictationController({ input, button, status, cancelButton, onResult: ({ finalText, interimText }) => { if (!valid(capture)) return; capture.finalText = finalText; + capture.reporter.event('revision', { text_length: (finalText + interimText).length, final: !interimText }); status.textContent = [finalText, interimText].filter(Boolean).join(' ') || 'Listening…'; }, - onError: error => { capture.failed = true; fail(capture, microphoneError(error)); }, + onError: error => { capture.failed = true; fail(capture, microphoneError(error), error?.error || error?.name); }, onEnd: () => { if (!capture.failed) commit(capture, capture.finalText || ''); } }); } else { @@ -144,6 +180,9 @@ export function createDictationController({ input, button, status, cancelButton, if (!valid(capture)) { release(capture); return; } const mimeType = recordingMimeType(root); capture.recorder = new root.MediaRecorder(capture.stream, mimeType ? { mimeType } : undefined); + const settings = capture.stream.getAudioTracks?.()[0]?.getSettings?.() || {}; + capture.reporter.event('started', { engine: 'recorded', format: capture.recorder.mimeType || mimeType, + sample_rate: settings.sampleRate, channel_count: settings.channelCount }); capture.recorder.ondataavailable = event => { if (!valid(capture) || !event.data?.size) return; capture.chunks.push(event.data); @@ -170,16 +209,21 @@ export function createDictationController({ input, button, status, cancelButton, })); capture.recorder.start(1000); } - if (valid(capture) && state !== 'error') render('recording', 'Listening. Select Finish dictation when you are ready.'); + if (!capture.live && valid(capture) && state !== 'error') render('recording', 'Listening. Select Finish dictation when you are ready.'); } catch (error) { - fail(capture, microphoneError(error)); + fail(capture, microphoneError(error), error?.name || 'input_failed'); } } async function finish() { - if (!current) return true; if (state === 'error') return false; + if (!current) return true; const capture = current; + if (capture.live) { + try { const text = await capture.live.finish(); if (text !== null) commit(capture, text); } + catch (_) { /* onState exposes the error and retains completed text. */ } + return capture.completion; + } if (capture.recorder?.state === 'recording') capture.recorder.stop(); capture.recognition?.stop(); if (state === 'requesting') { cancel(); return false; } @@ -190,11 +234,15 @@ export function createDictationController({ input, button, status, cancelButton, try { const response = await fetchImpl('/api/speech/capabilities', { credentials: 'same-origin', cache: 'no-store' }); const data = await response.json(); - capability = response.ok ? data.transcription : { available: false, reason: 'Sign in to use recorded transcription.' }; + capability = response.ok ? { ...data.transcription, streaming: data.streaming } : { available: false, reason: 'Sign in to use recorded transcription.' }; } catch (_) { capability = { available: false, reason: 'Unable to check recorded transcription. Try again when connected.' }; } if (!disposed && !current) { + if (!engineChosen && capability?.streaming?.available && engineSelect.querySelector('option[value="streaming"]')) { + engineSelect.value = 'streaming'; + engineChosen = true; + } render('idle', capability?.available ? 'Recorded audio and visible conversation context are sent to OpenAI for transcription. Audio is not saved by Von.' : capability?.reason || 'Recorded transcription is unavailable.'); @@ -207,6 +255,8 @@ export function createDictationController({ input, button, status, cancelButton, void transcribe(current); } }; const visibility = () => { if (root.document?.hidden && current) void finish(); }; + const chooseEngine = () => { engineChosen = true; }; + engineSelect.addEventListener('change', chooseEngine); button.addEventListener('click', click); cancelButton.addEventListener('click', cancel); retryButton.addEventListener('click', retry); @@ -216,8 +266,10 @@ export function createDictationController({ input, button, status, cancelButton, root.document?.addEventListener('visibilitychange', visibility); render('idle'); void refreshCapabilities(); - return { cancel, finish, start, refreshCapabilities, getState: () => state, + return { cancel, finish, start, refreshCapabilities, getAttemptIds: () => [...attemptIds], clearAttemptIds: () => { attemptIds.length = 0; }, + getState: () => state, hasPendingInput: () => !!current, dispose: () => { cancel(); disposed = true; + engineSelect.removeEventListener('change', chooseEngine); button.removeEventListener('click', click); cancelButton.removeEventListener('click', cancel); retryButton.removeEventListener('click', retry); root.removeEventListener?.('pagehide', cancel); root.removeEventListener?.('focus', refreshCapabilities); diff --git a/src/frontend/web/von_interface/static/js/streamingSpeech.js b/src/frontend/web/von_interface/static/js/streamingSpeech.js new file mode 100644 index 000000000..78f6e8e17 --- /dev/null +++ b/src/frontend/web/von_interface/static/js/streamingSpeech.js @@ -0,0 +1,247 @@ +// WebRTC only transports input audio. Von's existing turn runtime owns replies/tools. +export function createStreamingInput({ context, fetchImpl, onPartial, onFinal, onSpeechStart, + onState = () => {}, onEvent = () => {}, root = globalThis, audioContext: suppliedAudioContext }) { + let peer, channel, stream, cancelled = false, finishing = false, speaking = false; + let resolveFinish, rejectFinish, connectTimer; + const abort = new AbortController(); + const items = new Map(); + const order = []; + let emitted = 0, finishAcknowledged = false; + let model = null; + let audioContext = suppliedAudioContext, source, analyser, levelTimer, finishTimer; + let voiceSince = null, lastVoiceAt = null, noise = 0.003; + const completion = new Promise((resolve, reject) => { resolveFinish = resolve; rejectFinish = reject; }); + // Completion is also observed by callers that finish later. + completion.catch(() => {}); + function clean() { + root.clearTimeout?.(connectTimer); + root.clearTimeout?.(finishTimer); + root.clearInterval?.(levelTimer); + source?.disconnect(); analyser?.disconnect(); + if (!suppliedAudioContext) void audioContext?.close()?.catch?.(() => {}); + stream?.getTracks().forEach(track => track.stop()); + channel?.close(); peer?.close(); abort.abort(); + } + function cancel() { cancelled = true; clean(); resolveFinish(null); } + function fail(message) { + if (cancelled) return; + cancelled = true; clean(); onState('error', message); rejectFinish(new Error(message)); + } + function settled() { + if (finishing && finishAcknowledged && !speaking && [...items.values()].every(item => item.final)) { + const text = order.map(id => items.get(id)?.text || '').join(' ').trim(); + cancelled = true; clean(); resolveFinish(text); + } + } + function commitAudio() { + if (!cancelled && channel?.readyState === 'open') channel.send(JSON.stringify({ type: 'input_audio_buffer.commit' })); + } + function monitorPauses() { + source = audioContext.createMediaStreamSource(stream); + analyser = audioContext.createAnalyser(); analyser.fftSize = 1024; + source.connect(analyser); + const samples = new Float32Array(analyser.fftSize); + levelTimer = root.setInterval(() => { + if (cancelled || finishing || channel?.readyState !== 'open') return; + analyser.getFloatTimeDomainData(samples); + let power = 0; + for (const sample of samples) power += sample * sample; + const level = Math.sqrt(power / samples.length); + const now = root.performance.now(); + // A short sustained onset rejects isolated clicks. Browser capture + // supplies echo/noise cancellation; this detects acoustic pauses, + // not semantic completion. Keep explicit Finish/End available. + if (level >= Math.max(0.012, noise * 3)) { + voiceSince ??= now; + lastVoiceAt = now; + if (!speaking && now - voiceSince >= 120) { + speaking = true; onSpeechStart?.(); onEvent('speech_started'); + } + } else { + voiceSince = null; + if (!speaking) noise = noise * 0.98 + Math.min(level, 0.012) * 0.02; + if (speaking && now - lastVoiceAt >= 900) { + speaking = false; onEvent('speech_stopped'); commitAudio(); + } + } + }, 30); + } + function receive(event) { + if (cancelled) return; + const id = event.item_id; + if (event.type === 'error') { + if (event.error?.code === 'input_audio_buffer_commit_empty') { speaking = false; finishAcknowledged = true; settled(); return; } + fail('Live transcription stopped. Your completed text is preserved; try recorded dictation.'); return; + } + if (event.type === 'input_audio_buffer.speech_started') { + speaking = true; + if (id && !items.has(id)) items.set(id, { text: '', final: false }); + onSpeechStart?.(); onEvent('speech_started', { item_id: id }); + } + if (event.type === 'input_audio_buffer.speech_stopped') { + speaking = false; onEvent('speech_stopped', { item_id: id }); + } + if (event.type === 'input_audio_buffer.committed' && finishing) { finishAcknowledged = true; speaking = false; } + if (event.type === 'input_audio_buffer.committed' && id && !order.includes(id)) { + // Commit events are ordered; completion events can arrive out of order. + order.push(id); + if (!items.has(id)) items.set(id, { text: '', final: false }); + } + if (event.type === 'conversation.item.input_audio_transcription.delta' && id) { + const item = items.get(id) || { text: '', final: false }; + if (!item.final) item.text += event.delta || ''; + items.set(id, item); + onPartial?.(item.text); + } + if (event.type === 'conversation.item.input_audio_transcription.completed' && id) { + const item = items.get(id) || {}; + if (!item.final) { + items.set(id, { ...item, text: event.transcript || '', final: true }); + onEvent('committed', { item_id: id, text_length: (event.transcript || '').length, final: true }); + } + } + if (event.type === 'conversation.item.input_audio_transcription.failed') { + fail('An utterance could not be transcribed. Completed text is preserved.'); return; + } + while (emitted < order.length && items.get(order[emitted])?.final) { + const itemId = order[emitted++]; + onFinal?.(items.get(itemId).text, itemId); + } + settled(); + } + async function start() { + if (typeof root.RTCPeerConnection !== 'function' || !root.navigator?.mediaDevices?.getUserMedia || root.isSecureContext === false) { + fail('Live speech requires HTTPS and WebRTC. Choose recorded dictation.'); + return; + } + onState('requesting', 'Opening microphone…'); + try { + audioContext ||= new (root.AudioContext || root.webkitAudioContext)(); + await audioContext.resume(); + if (cancelled) return; + if (audioContext.state !== 'running') throw new Error('Tap the speech button to enable microphone processing.'); + stream = await root.navigator.mediaDevices.getUserMedia({ audio: { echoCancellation: true, noiseSuppression: true }, video: false }); + if (cancelled) { clean(); return; } + const settings = stream.getAudioTracks?.()[0]?.getSettings?.() || {}; + onEvent('started', { engine: 'webrtc', sample_rate: settings.sampleRate, channel_count: settings.channelCount }); + peer = new root.RTCPeerConnection(); + stream.getTracks().forEach(track => { + peer.addTrack(track, stream); + track.addEventListener?.('ended', () => { if (!cancelled && !finishing) fail('Microphone disconnected. Start speech again when ready.'); }); + }); + channel = peer.createDataChannel('oai-events'); + monitorPauses(); + channel.onmessage = ({ data }) => { try { receive(JSON.parse(data)); } catch (_) { fail('Invalid speech event. Reconnect to continue.'); } }; + channel.onopen = () => { root.clearTimeout?.(connectTimer); if (!cancelled) onState('listening', 'Listening…'); }; + channel.onclose = () => { if (!cancelled) fail('Speech disconnected. Completed text is preserved.'); }; + peer.onconnectionstatechange = () => { + if (!cancelled && ['failed', 'disconnected'].includes(peer.connectionState)) fail('Speech connection lost. Reconnect to continue.'); + }; + onState('connecting', 'Connecting live transcription…'); + const offer = await peer.createOffer(); + if (cancelled) return; + await peer.setLocalDescription(offer); + const response = await fetchImpl('/api/speech/connection', { + method: 'POST', credentials: 'same-origin', signal: abort.signal, + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ sdp: offer.sdp, context: context.text || '', vocabulary: context.vocabulary || [], language: context.language || '' }) + }); + const data = await response.json(); + if (cancelled) return; + if (!response.ok) throw new Error(data.message || 'Live speech is unavailable. Choose recorded dictation.'); + model = data.model; + onEvent('started', { engine: 'webrtc', model: data.model, vocabulary_count: data.vocabulary_count, context_chars: data.context_chars }); + await peer.setRemoteDescription({ type: 'answer', sdp: data.sdp }); + if (!cancelled && channel.readyState !== 'open') { + connectTimer = root.setTimeout?.(() => fail('The audio connection could not open. Try recorded dictation.'), 30000); + } + } catch (error) { + if (!cancelled) { fail(error.name === 'NotAllowedError' ? 'Microphone access was denied.' : error.message); throw error; } + } + } + function finish() { + if (cancelled || finishing) return completion; + finishing = true; + speaking = false; + stream?.getTracks().forEach(track => { track.enabled = false; }); + if (channel?.readyState !== 'open') { cancel(); return completion; } + onState('finishing', 'Finishing the current utterance…'); + commitAudio(); + // Release a paid media connection whose commit acknowledgement/final + // never arrives. Completed text remains recoverable through onState. + finishTimer = root.setTimeout?.(() => fail('The final transcript did not arrive. Completed text is preserved; reconnect to continue.'), 60000); + settled(); + return completion; + } + function updateContext(next) { + if (cancelled || finishing || channel?.readyState !== 'open') return; + const prompt = String(next.text || '').slice(-6000); + const keywords = [...new Set((next.vocabulary || []).filter(v => typeof v === 'string') + .map(v => v.replace(/[<>\r\n]/g, ' ').trim().slice(0, 100)).filter(Boolean))].slice(0, 80); + channel.send(JSON.stringify({ type: 'session.update', session: { type: 'transcription', audio: { input: { + transcription: { model, prompt, keywords } + } } } })); + onEvent('state', { state: 'context_updated', context_chars: prompt.length, vocabulary_count: keywords.length }); + } + return { start, finish, cancel, receive, updateContext, completedText: () => order.filter(id => items.get(id)?.final).map(id => items.get(id).text).join(' ') }; +} + +// PCM playback works across Web Audio browsers and starts before the HTTP stream +// is complete. Queue at most two seconds ahead so cancellation is prompt/bounded. +export function createSpeechPlayback({ fetchImpl, root = globalThis, onEvent = () => {} }) { + let audioContext, current = null; + function unlock() { + if (!audioContext) audioContext = new (root.AudioContext || root.webkitAudioContext)(); + return audioContext.resume(); + } + function stop(reason = 'interrupted') { + const old = current; current = null; + if (old) { + old.abort.abort(); old.nodes.forEach(node => { try { node.stop(); } catch (_) {} }); + onEvent('playback_ended', { reason }); + } + } + async function speak(text) { + stop('replaced'); + const playback = { abort: new AbortController(), nodes: new Set() }; + current = playback; + await unlock(); + if (current !== playback) return; + if (audioContext.state !== 'running') { stop('blocked'); throw new Error('Tap Start voice to enable audio playback.'); } + let time = audioContext.currentTime, tail = new Uint8Array(0), started = false; + try { + const response = await fetchImpl('/api/speech/speak', { method: 'POST', credentials: 'same-origin', signal: playback.abort.signal, + headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text }) }); + if (!response.ok) { + const error = await response.json().catch(() => ({})); + throw new Error(error.message || 'Speech could not play. The answer remains in chat.'); + } + if (!response.body?.getReader) throw new Error('Speech could not play. The answer remains in chat.'); + const reader = response.body.getReader(); + while (current === playback) { + const { done, value } = await reader.read(); + if (done) break; + const bytes = new Uint8Array(tail.length + value.length); bytes.set(tail); bytes.set(value, tail.length); + const count = Math.floor(bytes.length / 2); tail = bytes.slice(count * 2); + if (!count) continue; + const buffer = audioContext.createBuffer(1, count, 24000); + const samples = buffer.getChannelData(0), view = new DataView(bytes.buffer); + for (let i = 0; i < count; i++) samples[i] = view.getInt16(i * 2, true) / 32768; + while (current === playback && time - audioContext.currentTime > 2) await new Promise(resolve => root.setTimeout(resolve, 50)); + if (current !== playback) break; + const node = audioContext.createBufferSource(); node.buffer = buffer; node.connect(audioContext.destination); + playback.nodes.add(node); node.onended = () => playback.nodes.delete(node); + time = Math.max(time, audioContext.currentTime + 0.02); node.start(time); time += buffer.duration; + if (!started) { started = true; onEvent('playback_started', { + model: response.headers?.get?.('X-Speech-Model'), format: response.headers?.get?.('X-Speech-Format') + }); } + } + while (current === playback && playback.nodes.size) await new Promise(resolve => root.setTimeout(resolve, 50)); + if (current === playback) { current = null; onEvent('playback_ended', { reason: 'completed' }); } + } catch (error) { + if (current === playback) { stop('error'); throw error; } + } + } + function dispose() { stop('ended'); void audioContext?.close(); audioContext = null; } + return { unlock, speak, stop, dispose, getAudioContext: () => audioContext }; +} diff --git a/src/frontend/web/von_interface/static/js/voiceConversation.js b/src/frontend/web/von_interface/static/js/voiceConversation.js new file mode 100644 index 000000000..69cbe3cb7 --- /dev/null +++ b/src/frontend/web/von_interface/static/js/voiceConversation.js @@ -0,0 +1,102 @@ +import { createStreamingInput, createSpeechPlayback } from './streamingSpeech.js'; +import { createSpeechReporter } from './clientContext.js'; + +export function createVoiceConversation({ button, status, getContext, onSubmit, onStart = () => {}, + fetchImpl = (...args) => fetch(...args), root = globalThis }) { + let session = null, disposed = false, starting = false, generation = 0; + const playback = createSpeechPlayback({ fetchImpl, root, + onEvent: (name, values) => session?.reporter.event(name, { ...values, item_id: session.playbackItem }) }); + function valid(value) { + if (value === session && getContext().key !== value.context.key) end('Conversation changed. Voice ended.'); + return !disposed && session === value; + } + function end(message = 'Voice ended. Microphone released.') { + generation++; starting = false; + const old = session; + old?.reporter.event('ended', { reason: 'ended' }); + old?.input.cancel(); playback.stop('ended'); + session = null; + button.textContent = 'Start voice'; button.setAttribute('aria-pressed', 'false'); + status.textContent = message; + } + async function start() { + if (session || starting) return end(); + starting = true; const token = ++generation; + button.textContent = 'End voice'; button.setAttribute('aria-pressed', 'true'); + // Resume audio in the initiating gesture, before permission/network awaits. + try { await playback.unlock(); } catch (_) { if (generation === token) end('This browser could not open audio playback.'); return; } + if (generation !== token || disposed) return; + try { await onStart(); } + catch (error) { if (generation === token) end(error.message); return; } + if (disposed || generation !== token) return; + starting = false; + const context = getContext(); + const value = { context, submitted: new Set(), latestItem: null, heard: new Set() }; + value.reporter = createSpeechReporter({ fetchImpl, getContext: () => value.context, root }); + value.input = createStreamingInput({ context, fetchImpl, root, audioContext: playback.getAudioContext?.(), + onState: (state, message) => { + if (!valid(value)) return; + status.textContent = message; + value.reporter.event('state', { state }); + if (state === 'error') end(message + ' Select Start voice to reconnect, or use Dictate.'); + }, + onPartial: text => { if (valid(value)) status.textContent = text; }, + onSpeechStart: () => { + if (!valid(value)) return; + // Invalidate late spoken replies; already-running tools retain + // their canonical results and are never blindly repeated. + value.latestItem = null; + playback.stop('barge_in'); + value.reporter.event('interrupted', { reason: 'user_speech' }); + }, + onEvent: (name, values) => value.reporter.event(name, values), + onFinal: (text, itemId) => { + if (!valid(value) || !text.trim() || value.submitted.has(itemId)) return; + value.submitted.add(itemId); value.latestItem = itemId; + status.textContent = 'Von is responding. You can keep speaking.'; + void onSubmit(text, { + attemptId: value.reporter.attemptId, itemId, + submissionId: value.reporter.attemptId + '-' + itemId + }).then(() => { + value.reporter.event('submitted', { item_id: itemId, submission_id: value.reporter.attemptId + '-' + itemId }); + }).catch(() => { + if (valid(value)) end('Your message remains in the conversation queue for recovery. Voice paused.'); + }); + } + }); + session = value; + button.textContent = 'End voice'; button.setAttribute('aria-pressed', 'true'); + status.textContent = 'Opening voice. Audio is sent to OpenAI; Von’s voice is AI-generated.'; + try { await value.input.start(); } + catch (_) { /* Input exposes a recoverable error through onState. */ } + } + async function reply({ text, turnId, clientContext }) { + const value = session; + if (!value || !valid(value) || !text?.trim() || value.heard.has(turnId)) return false; + // Only the reply to this session's latest submitted utterance may speak. + if (!clientContext?.speech_attempt_ids?.includes(value.reporter.attemptId) + || clientContext.speech_item_id !== value.latestItem) return false; + value.heard.add(turnId); + value.playbackItem = value.latestItem; + value.input.updateContext?.(getContext()); + status.textContent = 'Von is speaking. Speak to interrupt.'; + try { await playback.speak(text); if (valid(value)) status.textContent = 'Listening…'; } + catch (error) { if (valid(value)) status.textContent = error.message; } + return true; + } + const click = () => { if (session || starting) end(); else void start(); }; + const leave = () => end(); + const hide = () => { if (root.document?.hidden && (session || starting)) end('Voice paused while the page is hidden. Start voice to resume.'); }; + const shortcut = event => { + if (event.altKey && event.shiftKey && event.code === 'KeyV') { event.preventDefault(); click(); } + }; + button.addEventListener('click', click); + root.document?.addEventListener('visibilitychange', hide); + root.addEventListener?.('pagehide', leave); + root.document?.addEventListener('keydown', shortcut); + return { start, end, reply, isActive: () => !!session, + dispose() { end(); disposed = true; playback.dispose(); button.removeEventListener('click', click); + root.document?.removeEventListener('visibilitychange', hide); root.removeEventListener?.('pagehide', leave); + root.document?.removeEventListener('keydown', shortcut); } + }; +} diff --git a/src/frontend/web/von_interface/templates/chat_tab.html b/src/frontend/web/von_interface/templates/chat_tab.html index ecde9d433..b92129e5f 100644 --- a/src/frontend/web/von_interface/templates/chat_tab.html +++ b/src/frontend/web/von_interface/templates/chat_tab.html @@ -246,6 +246,7 @@

Dictate + @@ -255,7 +256,8 @@

'); + const calls = []; + global.fetch = jest.fn(async (url, options = {}) => { + calls.push({ url, options }); + const payload = JSON.parse(options.body || '{}'); + const item = { ...payload, queue_id: 'voice-queue', status: 'queued' }; + return { ok: true, status: 200, json: async () => ({ success: true, item, items: [], history: [] }) }; + }); + require('../../src/frontend/web/von_interface/static/js/apiService.js').fetchWithTimeout + .mockImplementation((...args) => global.fetch(...args)); + chat.initializeChatTab(); + const { createVoiceConversation } = require('../../src/frontend/web/von_interface/static/js/voiceConversation.js'); + const options = createVoiceConversation.mock.calls.at(-1)[0]; + await options.onSubmit('Discuss Vontology', { attemptId: 'speech-attempt', itemId: 'item-one', submissionId: 'speech-attempt-item-one' }); + const enqueue = calls.find(c => c.url === '/von/api/chat_prompt_queue' && c.options.method === 'POST'); + const body = JSON.parse(enqueue.options.body); + expect(body).toMatchObject({ prompt_raw: 'Discuss Vontology', dispatch_mode: 'server', enqueue_submission_id: 'speech-attempt-item-one', + execution_envelope: { client_context: { speech_attempt_ids: ['speech-attempt'], speech_item_id: 'item-one' } } }); + expect(calls.some(c => String(c.url).startsWith('/von/generate'))).toBe(false); + }); + test('sends the stored Gemini choice as a bare model with its exact provider', async () => { const { getUserContext } = require('../../src/frontend/web/von_interface/static/js/apiService.js'); const { sendMessage } = require(chatTabModulePath); @@ -1182,9 +1209,10 @@ describe('chat task queue', () => { await new Promise((resolve) => setTimeout(resolve, 20)); await flushMicrotasks(); + expect(JSON.parse(global.fetch.mock.calls.find(([url]) => String(url).endsWith('/requeue'))[1].body).execution_envelope).toBeTruthy(); expect(generateBodies).toHaveLength(0); expect(document.querySelector('.chat-task-queue-item-label')?.textContent) - .toBe('Next up • Recovered'); + .toBe('Ready to resume • Recovered'); expect(document.querySelector('.chat-task-queue-item')).toBeTruthy(); expect(global.fetch.mock.calls.some(([url]) => String(url).endsWith('/claim'))).toBe(false); expect(global.fetch.mock.calls.some(([url]) => String(url).endsWith('/finish'))).toBe(false); @@ -1238,7 +1266,7 @@ describe('chat task queue', () => { expect(document.getElementById('chatTaskQueueCount')?.textContent).toBe('1 queued'); const labels = Array.from(document.querySelectorAll('.chat-task-queue-item-label')) .map((node) => node.textContent); - expect(labels).toEqual(['Next up • Current']); + expect(labels).toEqual(['Ready to resume • Current']); expect(document.querySelector('.chat-task-queue-item-running')).toBeNull(); expect(document.querySelector( '#chatSessionTabs [data-session-id="session-2"] .chat-session-tab-activity' diff --git a/tests/frontend/dictation.test.js b/tests/frontend/dictation.test.js index fa1dda5d3..9d3dc5520 100644 --- a/tests/frontend/dictation.test.js +++ b/tests/frontend/dictation.test.js @@ -41,7 +41,7 @@ describe('recorded dictation lifecycle', () => { expect(await finish).toBe(true); expect(input.value).toBe('Edited while waiting Vontology works.'); expect(stopTrack).toHaveBeenCalled(); - const form = fetchImpl.mock.calls[1][1].body; + const form = fetchImpl.mock.calls.find(([url]) => url.endsWith('/transcribe'))[1].body; expect(form.get('audio').type).toBe('audio/mp4'); expect(form.get('vocabulary')).toContain('Wikidata'); expect(form.get('language')).toBe('en-NZ'); @@ -81,6 +81,7 @@ describe('recorded dictation lifecycle', () => { await controller.start(); expect(document.querySelector('p').textContent).toContain('Microphone access was denied'); expect(await controller.finish()).toBe(false); + expect(controller.hasPendingInput()).toBe(false); }); test('inserts at selected range only when original draft is unchanged', () => { expect(insertDictation('Hello old world', 'new', { start: 6, end: 9 })).toBe('Hello new world'); diff --git a/tests/frontend/streamingSpeech.test.js b/tests/frontend/streamingSpeech.test.js new file mode 100644 index 000000000..2dd5308ef --- /dev/null +++ b/tests/frontend/streamingSpeech.test.js @@ -0,0 +1,95 @@ +/** @jest-environment jsdom */ +const { createStreamingInput, createSpeechPlayback } = require('../../src/frontend/web/von_interface/static/js/streamingSpeech.js'); +const flush = async () => { for (let i = 0; i < 15; i++) await Promise.resolve(); }; +function setup(overrides = {}) { + let peer; + const track = { enabled: true, stop: jest.fn(), addEventListener: jest.fn() }; + const channel = { readyState: 'open', send: jest.fn(), close: jest.fn() }; + class Peer { + constructor() { peer = this; } + addTrack() {} + createDataChannel() { return channel; } + async createOffer() { return { sdp: 'v=0\r\nfixture' }; } + async setLocalDescription() {} + async setRemoteDescription() { channel.onopen(); } + close() {} + } + class AudioContext { + state = 'running'; + async resume() {} + createMediaStreamSource() { return { connect() {}, disconnect() {} }; } + createAnalyser() { return { fftSize: 1024, getFloatTimeDomainData() {}, disconnect() {} }; } + close() {} + } + const root = { navigator: { mediaDevices: { getUserMedia: jest.fn(async () => ({ getTracks: () => [track] })) } }, + RTCPeerConnection: Peer, AudioContext, setTimeout, clearTimeout, setInterval, clearInterval, performance, isSecureContext: true, ...overrides }; + const onFinal = jest.fn(), onPartial = jest.fn(), onState = jest.fn(); + const fetchImpl = jest.fn(async () => ({ ok: true, json: async () => ({ sdp: 'v=0\r\nanswer', model: 'gpt-live-transcribe' }) })); + const input = createStreamingInput({ context: { text: 'Vontology', vocabulary: ['Von'], language: 'en-NZ' }, root, fetchImpl, onFinal, onPartial, onState }); + return { input, root, track, channel, onFinal, onPartial, onState, fetchImpl, peer: () => peer }; +} +const event = (input, type, item_id, fields = {}) => input.receive({ type, item_id, ...fields }); +test('reordered completions and duplicate finals emit exactly once in committed order', async () => { + const { input, onFinal } = setup(); await input.start(); + event(input, 'input_audio_buffer.committed', 'one'); event(input, 'input_audio_buffer.committed', 'two'); + event(input, 'conversation.item.input_audio_transcription.completed', 'two', { transcript: 'second' }); + expect(onFinal).not.toHaveBeenCalled(); + event(input, 'conversation.item.input_audio_transcription.completed', 'one', { transcript: 'first' }); + event(input, 'conversation.item.input_audio_transcription.completed', 'one', { transcript: 'first' }); + expect(onFinal.mock.calls).toEqual([['first', 'one'], ['second', 'two']]); input.cancel(); +}); +test('finish waits for commit acknowledgement and final transcript, then stops tracks', async () => { + const { input, channel, track } = setup(); await input.start(); + const finish = input.finish(); let finished = false; finish.then(() => { finished = true; }); + await flush(); expect(finished).toBe(false); + expect(JSON.parse(channel.send.mock.calls[0][0]).type).toBe('input_audio_buffer.commit'); + event(input, 'input_audio_buffer.committed', 'one'); + event(input, 'conversation.item.input_audio_transcription.completed', 'one', { transcript: 'Von' }); + expect(await finish).toBe('Von'); expect(track.stop).toHaveBeenCalled(); +}); +test('cancel while permission is outstanding releases a late microphone and makes no provider call', async () => { + let grant; const permission = new Promise(resolve => { grant = resolve; }); + const { input, root, track, fetchImpl } = setup(); root.navigator.mediaDevices.getUserMedia.mockReturnValue(permission); + const start = input.start(); await flush(); input.cancel(); grant({ getTracks: () => [track] }); await start; + expect(track.stop).toHaveBeenCalled(); expect(fetchImpl).not.toHaveBeenCalled(); +}); +test('transport failure closes microphone and prevents late transcript callbacks', async () => { + const { input, track, onFinal, peer } = setup(); await input.start(); + peer().connectionState = 'failed'; peer().onconnectionstatechange(); + event(input, 'input_audio_buffer.committed', 'late'); + event(input, 'conversation.item.input_audio_transcription.completed', 'late', { transcript: 'late' }); + expect(track.stop).toHaveBeenCalled(); expect(onFinal).not.toHaveBeenCalled(); +}); +test('cancel during output unlock cannot issue a late synthesis request', async () => { + let unlock; + const resume = new Promise(resolve => { unlock = resolve; }); + class AudioContext { constructor() { this.state = 'running'; } resume() { return resume; } close() {} } + const fetchImpl = jest.fn(); + const player = createSpeechPlayback({ fetchImpl, root: { AudioContext } }); + const speaking = player.speak('A response'); player.stop(); unlock(); await speaking; + expect(fetchImpl).not.toHaveBeenCalled(); player.dispose(); +}); + +test('short gaps and isolated clicks do not commit; a sustained utterance followed by a pause commits once', async () => { + jest.useFakeTimers(); + let level = 0; + const analyser = { fftSize: 1024, getFloatTimeDomainData: samples => samples.fill(level), disconnect() {} }; + class AudioContext { + state = 'running'; + async resume() {} + createMediaStreamSource() { return { connect() {}, disconnect() {} }; } + createAnalyser() { return analyser; } + close() {} + } + const { input, channel } = setup({ AudioContext, performance: { now: () => Date.now() } }); + try { + await input.start(); + level = 0.1; jest.advanceTimersByTime(60); level = 0; jest.advanceTimersByTime(1200); + expect(channel.send).not.toHaveBeenCalled(); + level = 0.1; jest.advanceTimersByTime(600); level = 0; jest.advanceTimersByTime(300); + level = 0.1; jest.advanceTimersByTime(500); + expect(channel.send).not.toHaveBeenCalled(); + level = 0; jest.advanceTimersByTime(1100); jest.advanceTimersByTime(2000); + expect(channel.send).toHaveBeenCalledTimes(1); + } finally { input.cancel(); jest.useRealTimers(); } +}); diff --git a/tests/frontend/voiceConversation.test.js b/tests/frontend/voiceConversation.test.js new file mode 100644 index 000000000..5a075d0a0 --- /dev/null +++ b/tests/frontend/voiceConversation.test.js @@ -0,0 +1,46 @@ +/** @jest-environment jsdom */ +jest.mock('../../src/frontend/web/von_interface/static/js/streamingSpeech.js', () => ({ + createStreamingInput: jest.fn(), createSpeechPlayback: jest.fn() +})); +const media = require('../../src/frontend/web/von_interface/static/js/streamingSpeech.js'); +const { createVoiceConversation } = require('../../src/frontend/web/von_interface/static/js/voiceConversation.js'); +const flush = async () => { for (let i = 0; i < 15; i++) await Promise.resolve(); }; +let voice, input, callbacks, playback, context, submit; +beforeEach(() => { + jest.clearAllMocks(); + document.body.innerHTML = '

'; + input = { start: jest.fn(async () => {}), cancel: jest.fn() }; + playback = { unlock: jest.fn(async () => {}), speak: jest.fn(async () => {}), stop: jest.fn(), dispose: jest.fn() }; + media.createStreamingInput.mockImplementation(options => { callbacks = options; return input; }); + media.createSpeechPlayback.mockReturnValue(playback); + context = { key: 'actor:conversation', sessionId: 'conversation' }; + submit = jest.fn(async () => {}); + voice = createVoiceConversation({ button: document.querySelector('button'), status: document.querySelector('p'), + getContext: () => context, onSubmit: submit, fetchImpl: jest.fn(async () => ({ ok: true })), root: window }); +}); +afterEach(() => voice.dispose()); +test('duplicate final produces one submission; only latest utterance reply speaks', async () => { + await voice.start(); callbacks.onFinal('First', 'one'); callbacks.onFinal('First', 'one'); await flush(); + expect(submit).toHaveBeenCalledTimes(1); + const { attemptId } = submit.mock.calls[0][1]; + await voice.reply({ text: 'Answer', turnId: 'turn1', clientContext: { speech_attempt_ids: [attemptId], speech_item_id: 'one' } }); + expect(playback.speak).toHaveBeenCalledTimes(1); + callbacks.onSpeechStart(); + expect(playback.stop).toHaveBeenCalledWith('barge_in'); + await voice.reply({ text: 'Old answer', turnId: 'late', clientContext: { speech_attempt_ids: [attemptId], speech_item_id: 'one' } }); + expect(playback.speak).toHaveBeenCalledTimes(1); +}); +test('End cancels capture and playback; late final cannot submit', async () => { + await voice.start(); voice.end(); callbacks.onFinal('late', 'one'); + expect(input.cancel).toHaveBeenCalled(); expect(playback.stop).toHaveBeenCalledWith('ended'); + expect(submit).not.toHaveBeenCalled(); +}); +test('conversation switch ends capture before next final can write elsewhere', async () => { + await voice.start(); context = { key: 'other', sessionId: 'other' }; callbacks.onFinal('wrong scope', 'one'); + expect(input.cancel).toHaveBeenCalled(); expect(submit).not.toHaveBeenCalled(); +}); +test('End while audio unlock waits prevents late microphone opening', async () => { + let unlock; playback.unlock.mockReturnValue(new Promise(resolve => { unlock = resolve; })); + const start = voice.start(); voice.end(); unlock(); await start; + expect(media.createStreamingInput).not.toHaveBeenCalled(); +});