Fix main_v2 bridge and pipeline regressions

config.json

@@ -18,6 +18,11 @@
     "post_correct": false,
     "post_correct_llm": false,
     "post_correct_model": "qwen:2b",
+    "post_correct_ollama_url": "http://127.0.0.1:11434/api/generate",
+    "post_correct_llm_timeout": 8.0,
+    "post_correct_keep_alive": "30m",
+    "post_correct_min_overlap": 0.45,
+    "post_correct_debug": false,
     "llm_paragraph": false,
     "only_translate_llm": false,
     "temperature_fallback": [0.0, 0.2, 0.4, 0.6, 0.8, 1.0],

engine_distribute.py

@@ -7,6 +7,21 @@ import os
 
 INGEST_URL = "https://emiapi.reynafamily.com/live-captions/ingest"
 
+def post_with_retries(payload, timeout, max_attempts=3):
+    delay = 1
+    for attempt in range(1, max_attempts + 1):
+        try:
+            response = requests.post(INGEST_URL, json=payload, timeout=timeout)
+            if response.status_code == 200:
+                return True
+            print(f"[Distribute Error] {response.status_code} on attempt {attempt}/{max_attempts}.")
+        except Exception as exc:
+            print(f"[Distribute Error] {exc} on attempt {attempt}/{max_attempts}.")
+        if attempt < max_attempts:
+            time.sleep(delay)
+            delay = min(delay * 2, 5)
+    return False
+
 def log_debug(message):
     try:
         with open("distribute_debug.log", "a") as f:
@@ -31,8 +46,7 @@ def run_distribution(in_queue, args):
                 sys.stdout.write(f"\r\033[K[DRAFT]: {draft_text}")
                 sys.stdout.flush()
                 if args.ingest:
-                    try: requests.post(INGEST_URL, json={"draft": draft_text}, timeout=2)
-                    except: pass
+                    post_with_retries({"draft": draft_text}, timeout=2, max_attempts=2)
                 continue
 
             # Finalized segment
@@ -75,18 +89,9 @@ def run_distribution(in_queue, args):
                     should_send = False
                 
                 if should_send:
-                    delay, max_delay, success = 1, 15, False
-                    while not success:
-                        try:
-                            response = requests.post(INGEST_URL, json=ingest_payload, timeout=5)
-                            if response.status_code == 200: success = True
-                            else: print(f"[Distribute Error] {response.status_code}. Retrying...")
-                        except Exception as e:
-                            print(f"[Distribute Error] {e}. Retrying...")
-                        
-                        if not success:
-                            time.sleep(delay)
-                            delay = min(delay * 2, max_delay)
+                    success = post_with_retries(ingest_payload, timeout=5, max_attempts=3)
+                    if not success:
+                        log_debug(f"Dropped ingest payload after retries: {ingest_payload}")
                 
         except Exception as e:
             print(f"[Distribute] Error: {e}")

engine_llm.py

@@ -15,6 +15,7 @@ import time
 import argparse
 import multiprocessing
 import os
+import re
 import requests
 from collections import deque
 
@@ -26,18 +27,74 @@ def run_llm_processor(in_queue, out_queue, args):
     
     api_key = os.environ.get("OPENAI_API_KEY")
     is_openai = args.post_correct_model.startswith("gpt-")
-    target_lang = "English"
-    if args.lang and args.lang != 'en': target_lang = args.lang
+    llm_state = {"line_warned": False, "paragraph_warned": False}
+
+    def normalize_english_caption(text):
+        normalized = " ".join((text or "").split()).strip()
+        if normalized and normalized[0].isalpha():
+            normalized = normalized[0].upper() + normalized[1:]
+        return normalized
+
+    def apply_rule_based_post_correction(text):
+        corrected = normalize_english_caption(text)
+        corrected = re.sub(r"\s+", " ", corrected).strip()
+        corrected = re.sub(r"\s+([,.;!?])", r"\1", corrected)
+        corrected = re.sub(r"([,.;!?]){2,}", r"\1", corrected)
+        corrected = re.sub(r"\b(uh+|um+|erm+|ah+|hmm+)\b", "", corrected, flags=re.IGNORECASE)
+        corrected = re.sub(r"\s+", " ", corrected).strip()
+        corrected = re.sub(r"\b(\w+)\s+\1\b", r"\1", corrected, flags=re.IGNORECASE)
+        return normalize_english_caption(corrected)
+
+    def word_overlap_ratio(source_text, candidate_text):
+        src_tokens = set(re.findall(r"[a-z0-9']+", source_text.lower()))
+        cand_tokens = set(re.findall(r"[a-z0-9']+", candidate_text.lower()))
+        if not src_tokens:
+            return 1.0
+        return len(src_tokens & cand_tokens) / len(src_tokens)
+
+    def call_openai(messages):
+        response = requests.post(
+            "https://api.openai.com/v1/chat/completions",
+            json={"model": args.post_correct_model, "messages": messages, "temperature": 0.1},
+            headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+            timeout=getattr(args, "post_correct_llm_timeout", 8.0),
+        )
+        response.raise_for_status()
+        return response.json()["choices"][0]["message"]["content"].strip()
+
+    def call_ollama(prompt, temperature):
+        response = requests.post(
+            getattr(args, "post_correct_ollama_url", "http://127.0.0.1:11434/api/generate"),
+            json={
+                "model": args.post_correct_model,
+                "prompt": prompt,
+                "stream": False,
+                "options": {"temperature": temperature},
+                "keep_alive": getattr(args, "post_correct_keep_alive", "30m"),
+            },
+            timeout=getattr(args, "post_correct_llm_timeout", 8.0),
+        )
+        response.raise_for_status()
+        return response.json().get("response", "").strip()
+
+    def call_llm(messages, prompt, temperature, warn_key):
+        try:
+            if is_openai:
+                if not api_key:
+                    raise RuntimeError("OPENAI_API_KEY is not set")
+                return call_openai(messages)
+            return call_ollama(prompt, temperature)
+        except Exception as exc:
+            if not llm_state[warn_key]:
+                print(f"[LLM] {warn_key.replace('_', ' ').title()} unavailable ({exc}). Falling back to deterministic text.")
+                llm_state[warn_key] = True
+            return ""
 
     # State: This is the ONLY text we send to the LLM as context
     active_context = ""
+    recent_lines = deque(maxlen=2)
     
     def call_llm_rolling_refine(new_segments_str, context):
-        if not is_openai: return ""
-        
-        url = "https://api.openai.com/v1/chat/completions"
-        headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
-        
         prompt = f"""Task: Refine the following live transcription stream into clean, professional paragraphs.
 
 [PREVIOUS WORKING CONTEXT]
@@ -50,21 +107,50 @@ def run_llm_processor(in_queue, out_queue, args):
 1. INTEGRATE: Polished and merge the new segments into the flow of the 'PREVIOUS WORKING CONTEXT'.
 2. CONSOLIDATE: Remove redundant repetitions and translator echoes.
 3. ORGANIZE: Use a double newline (\\n\\n) to start a new paragraph when a topic changes or the current one is complete.
-4. TARGET LANGUAGE: Output ONLY in {target_lang}.
+4. TARGET LANGUAGE: Output ONLY in English.
 5. OUTPUT: Provide ONLY the refined, consolidated text. Do not explain anything.
 """
-        payload = {
-            "model": args.post_correct_model,
-            "messages": [{"role": "system", "content": "You are a live transcript editor."},
-                         {"role": "user", "content": prompt}],
-            "temperature": 0.1
-        }
-        try:
-            resp = requests.post(url, json=payload, headers=headers, timeout=20)
-            if resp.status_code == 200:
-                return resp.json()["choices"][0]["message"]["content"].strip()
-            return ""
-        except: return ""
+        messages = [
+            {"role": "system", "content": "You are a live transcript editor."},
+            {"role": "user", "content": prompt},
+        ]
+        return call_llm(messages, prompt, 0.1, "paragraph_warned")
+
+    def post_correct_line(text):
+        if not getattr(args, "post_correct", False):
+            return normalize_english_caption(text)
+
+        corrected = apply_rule_based_post_correction(text)
+        if not getattr(args, "post_correct_llm", False):
+            return corrected
+
+        prev2 = recent_lines[-2] if len(recent_lines) >= 2 else ""
+        prev1 = recent_lines[-1] if len(recent_lines) >= 1 else ""
+        prompt = (
+            "You are a real-time English caption corrector for live speech.\n"
+            "Task:\n"
+            "Clean ONLY the current caption line.\n"
+            "Hard rules:\n"
+            "1. Preserve meaning exactly. Never replace current content with prior context.\n"
+            "2. Remove disfluencies and false starts.\n"
+            "3. Fix punctuation, casing, and obvious STT typos.\n"
+            "4. If uncertain, return the original line unchanged.\n"
+            "5. Output ONLY one corrected English line.\n"
+            "Context (reference only):\n"
+            f"Previous line 1: {prev1}\n"
+            f"Previous line 2: {prev2}\n"
+            "Current line to correct:\n"
+            f"{corrected}\n"
+            "Corrected:"
+        )
+        messages = [
+            {"role": "system", "content": "You are a real-time English caption corrector."},
+            {"role": "user", "content": prompt},
+        ]
+        candidate = normalize_english_caption(call_llm(messages, prompt, 0.1, "line_warned"))
+        if candidate and word_overlap_ratio(corrected, candidate) >= getattr(args, "post_correct_min_overlap", 0.45):
+            return candidate
+        return corrected
 
     pending_buffer = []
     last_llm_call_time = time.time()
@@ -81,15 +167,18 @@ def run_llm_processor(in_queue, out_queue, args):
             bridge_text = item.get("en_bridge", raw_text).strip()
             if not raw_text: continue
 
+            corrected_text = post_correct_line(bridge_text)
+
             # Hallucination check
-            words = bridge_text.lower().split()
+            words = corrected_text.lower().split()
             if len(words) > 10 and len(set(words)) < 3: continue
 
-            pending_buffer.append(bridge_text)
+            pending_buffer.append(corrected_text)
+            recent_lines.append(corrected_text)
 
             should_refine = False
             if len(pending_buffer) >= 5: should_refine = True
-            elif len(pending_buffer) >= 2 and bridge_text.endswith((".", "?", "!")): should_refine = True
+            elif len(pending_buffer) >= 2 and corrected_text.endswith((".", "?", "!")): should_refine = True
             elif len(pending_buffer) >= 1 and (time.time() - last_llm_call_time) > 15: should_refine = True
 
             structured_paragraph = None
@@ -110,13 +199,16 @@ def run_llm_processor(in_queue, out_queue, args):
                         # No break, just update the context
                         active_context = result
                         structured_paragraph = result
-                    
-                    pending_buffer = []
-                    last_llm_call_time = time.time()
+                else:
+                    structured_paragraph = new_batch
+                    active_context = new_batch
+
+                pending_buffer = []
+                last_llm_call_time = time.time()
 
             out_queue.put({
                 "raw": raw_text,
-                "corrected": bridge_text,
+                "corrected": corrected_text,
                 "paragraph": structured_paragraph,
                 "detected_lang": item.get("detected_lang"),
                 "speaker": item.get("speaker"),

engine_transcribe.py

@@ -100,7 +100,7 @@ def run_transcription(out_queue, args):
         if status: print(status, file=sys.stderr)
         audio_queue.put(indata.copy())
 
-    audio_buffer, speech_started, rolling_context = [], False, ""
+    audio_buffer, speech_started, rolling_context, rolling_context_en = [], False, "", ""
     last_stream_time = time.time()
     
     try:
@@ -127,7 +127,7 @@ def run_transcription(out_queue, args):
                             transcribe_kwargs = {
                                 "path_or_hf_repo": model_name,
                                 "temperature": args.temperature_fallback,
-                                "word_timestamps": True,
+                                "word_timestamps": args.speaker_diarization,
                                 "logprob_threshold": args.logprob_threshold,
                                 "compression_ratio_threshold": args.compression_threshold,
                             }
@@ -139,13 +139,26 @@ def run_transcription(out_queue, args):
                             detected_lang = res_asr.get('language', args.lang or 'en')
 
                             if text:
+                                if args.filter_lang and args.lang and detected_lang != args.lang:
+                                    print(f"[Transcribe] Discarding segment (Detected: {detected_lang}, Expected: {args.lang})")
+                                    audio_buffer, speech_started = [], False
+                                    continue
+
                                 english_text = text
-                                if detected_lang != "en" or args.lang == 'en':
-                                    if detected_lang != "en":
-                                        bridge_kwargs = transcribe_kwargs.copy()
-                                        bridge_kwargs["task"] = "translate"
-                                        res_bridge = transcribe_with_controls(current_audio, bridge_kwargs)
-                                        english_text = res_bridge['text'].strip()
+                                if detected_lang != "en":
+                                    bridge_kwargs = {
+                                        "path_or_hf_repo": model_name,
+                                        "task": "translate",
+                                        "temperature": args.temperature_fallback,
+                                        "logprob_threshold": args.logprob_threshold,
+                                        "compression_ratio_threshold": args.compression_threshold,
+                                    }
+                                    if args.lang:
+                                        bridge_kwargs["language"] = args.lang
+                                    if args.context and rolling_context_en:
+                                        bridge_kwargs["initial_prompt"] = rolling_context_en
+                                    res_bridge = transcribe_with_controls(current_audio, bridge_kwargs)
+                                    english_text = res_bridge.get("text", "").strip() or text
 
                                 speaker_label = None
                                 if diarization_pipeline:
@@ -171,7 +184,9 @@ def run_transcription(out_queue, args):
                                 
                                 speaker_tag = f" ({speaker_label})" if speaker_label else ""
                                 print(f"[Transcribe] {detected_lang.upper()}{speaker_tag}: {text}")
-                                if args.context: rolling_context = (rolling_context + " " + text)[-200:].strip()
+                                if args.context:
+                                    rolling_context = (rolling_context + " " + text)[-200:].strip()
+                                    rolling_context_en = (rolling_context_en + " " + english_text)[-200:].strip()
 
                             audio_buffer, speech_started = [], False
                         

engine_translate.py

@@ -85,7 +85,7 @@ def run_translation(in_queue, out_queue, args):
             else:
                 text_to_translate = paragraph_text or corrected_text or raw_text
                 if args.en:
-                    payload["en"] = corrected_text or raw_text
+                    payload["en"] = text_to_translate
 
             if text_to_translate and translation_engines:
                 for lang_key, (model, tokenizer) in translation_engines.items():

history.md

@@ -238,3 +238,20 @@ The project now features a high-performance, Apple Silicon-optimized pipeline th
     - **English Bridge Optimization:** Forced a two-pass transcription strategy (ASR first, then optional translation) to ensure word-level timestamps are captured for speaker mapping even when translating to English.
     - **Stability Fixes:** Added hallucination filtering for Whisper's repetitive loops, enforced line-buffering for terminal logging, and implemented a global `lzma` mock to ensure compatibility across restricted Python environments.
 - **Outcome:** A robust, production-ready pipeline that produces high-quality, speaker-attributed, and professionally formatted live transcripts.
+
+## Phase 28: `main_v2` Bridge Contract Fixes + Live Reliability Review
+- **Goal:** Repair stage-contract regressions in the new multi-process pipeline and preserve live throughput under failure.
+- **Review Findings:**
+    - The Whisper English bridge in `engine_transcribe.py` reused source-language prompt context for `task="translate"` instead of a dedicated English context, which could bias or degrade translated bridge text.
+    - `--filter-lang` was exposed in `main_v2.py` but not enforced in the new transcription engine, so wrong-language speech could still be bridged and propagated downstream.
+    - `engine_llm.py` told the paragraph refiner to output the source language even though downstream Marian models require English input, breaking `llm_paragraph` for non-English sources.
+    - `engine_translate.py` could translate one English variant while publishing a different `[EN]` line, making the visible source text diverge from what target-language MT actually used.
+    - `engine_distribute.py` retried ingest forever in the queue consumer, allowing a network outage to stall the whole live pipeline.
+    - `main_v2.py` exposed post-correction settings that the new pipeline did not fully honor, making CLI behavior drift from the documented workflow.
+- **Approach:**
+    - Restored a dedicated rolling English prompt context for the Whisper bridge pass and rebuilt bridge kwargs explicitly instead of copying source-ASR kwargs.
+    - Reinstated `--filter-lang` enforcement before bridge generation and reduced unnecessary ASR cost by only asking Whisper for word timestamps when diarization is enabled.
+    - Updated the LLM stage so paragraph refinement always preserves English bridge text, added deterministic post-correction plus optional LLM line cleanup, and provided a safe fallback paragraph when the LLM backend is unavailable.
+    - Aligned published English output with the exact text chosen for translation and added the missing Ollama-related config/plumbing to `main_v2`.
+    - Replaced infinite ingest retry loops with bounded retry logic so failed delivery degrades gracefully instead of freezing caption flow.
+- **Outcome:** The `main_v2` pipeline now keeps an English-first contract across transcription, refinement, and Marian translation while behaving more predictably under API outages and mixed-language input.

main_v2.py

@@ -50,6 +50,11 @@ def main():
     parser.add_argument("--post-correct", action="store_true", default=defaults.get("post_correct", False))
     parser.add_argument("--post-correct-llm", action="store_true", default=defaults.get("post_correct_llm", False))
     parser.add_argument("--post-correct-model", type=str, default=defaults.get("post_correct_model", "qwen:2b"))
+    parser.add_argument("--post-correct-ollama-url", type=str, default=defaults.get("post_correct_ollama_url", "http://127.0.0.1:11434/api/generate"))
+    parser.add_argument("--post-correct-llm-timeout", type=float, default=defaults.get("post_correct_llm_timeout", 8.0))
+    parser.add_argument("--post-correct-keep-alive", type=str, default=defaults.get("post_correct_keep_alive", "30m"))
+    parser.add_argument("--post-correct-min-overlap", type=float, default=defaults.get("post_correct_min_overlap", 0.45))
+    parser.add_argument("--post-correct-debug", action="store_true", default=defaults.get("post_correct_debug", False))
     parser.add_argument("--llm-paragraph", action="store_true", default=defaults.get("llm_paragraph", False))
 
     # Translate Args