fix: preserve stable realtime transcript on lock (#3408)

Co-authored-by: zhifu gao <[email protected]>

funasr/bin/realtime_ws.py

@@ -33,36 +33,110 @@ logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(me
 logger = logging.getLogger(__name__)
 
 
+def _normalize_transcript_with_positions(text):
+    normalized = []
+    source_positions = []
+    for index, char in enumerate(text):
+        if regex.fullmatch(r"[\p{P}\p{S}\s]", char):
+            continue
+        folded = char.casefold()
+        normalized.extend(folded)
+        source_positions.extend([index] * len(folded))
+    return "".join(normalized), source_positions
+
+
+def _normalize_transcript(text):
+    return _normalize_transcript_with_positions(text)[0]
+
+
+def _merge_overlapping_transcripts(existing, incoming):
+    """Join adjacent rolling-window transcripts when their text overlap is clear."""
+    existing_norm, existing_positions = _normalize_transcript_with_positions(existing)
+    incoming_norm, incoming_positions = _normalize_transcript_with_positions(incoming)
+    if not existing_norm or not incoming_norm:
+        return None
+
+    minimum_overlap = max(4, min(len(existing_norm), len(incoming_norm)) // 4)
+    for overlap in range(min(len(existing_norm), len(incoming_norm)), minimum_overlap - 1, -1):
+        if existing_norm[-overlap:] != incoming_norm[:overlap]:
+            continue
+        existing_start = len(existing_norm) - overlap
+        if (
+            existing_start > 0
+            and existing_positions[existing_start - 1] == existing_positions[existing_start]
+        ):
+            continue
+        if (
+            overlap < len(incoming_positions)
+            and incoming_positions[overlap - 1] == incoming_positions[overlap]
+        ):
+            continue
+        suffix_start = incoming_positions[overlap - 1] + 1
+        while (
+            suffix_start < len(incoming)
+            and regex.fullmatch(r"[\p{P}\p{S}]", incoming[suffix_start])
+            and existing.endswith(incoming[suffix_start])
+        ):
+            suffix_start += 1
+        return existing + incoming[suffix_start:]
+    return None
+
+
 def detect_and_fix_hallucination(text, max_ngram_length=12, max_occurrences=3):
     """Detect repeated patterns (hallucination) and truncate to keep one occurrence."""
     if not text or len(text) < max_ngram_length * 2:
         return text, False
 
-    cleaned = regex.sub(r'\p{P}+', '', text)
+    source_positions = []
+    cleaned_chars = []
+    for index, char in enumerate(text):
+        if regex.fullmatch(r'\p{P}', char):
+            continue
+        cleaned_chars.append(char)
+        source_positions.append(index)
+    cleaned = ''.join(cleaned_chars)
+
+    def truncate_after_first(match):
+        def ascii_quote_count(quote, end):
+            count = 0
+            for index in range(end):
+                if text[index] != quote:
+                    continue
+                if (
+                    quote == "'"
+                    and index > 0
+                    and index + 1 < len(text)
+                    and text[index - 1].isalnum()
+                    and text[index + 1].isalnum()
+                ):
+                    continue
+                count += 1
+            return count
+
+        cleaned_end = match.start(1) + len(match.group(1))
+        source_end = source_positions[cleaned_end - 1] + 1
+        while source_end < len(text) and regex.fullmatch(
+            r'[\p{Pe}\p{Pf}\p{Po}\p{Pd}\p{Pc}]', text[source_end]
+        ):
+            if (
+                text[source_end] in {'"', "'"}
+                and ascii_quote_count(text[source_end], source_end) % 2 == 0
+            ):
+                break
+            source_end += 1
+        return text[:source_end].rstrip(), True
 
     word_pattern = rf'(?<!\S)(?!\d+$)(\w+)(?:\s+\1){{{max_occurrences - 1},}}(?!\S)'
-    if regex.search(word_pattern, cleaned, regex.IGNORECASE):
-        match = regex.search(word_pattern, cleaned, regex.IGNORECASE)
-        repeated = match.group(1)
-        pos = text.find(repeated)
-        if pos >= 0:
-            end_pos = text.find(repeated, pos + len(repeated))
-            if end_pos >= 0:
-                return text[:end_pos + len(repeated)], True
-        return text[:len(text)//2], True
+    match = regex.search(word_pattern, cleaned, regex.IGNORECASE)
+    if match:
+        return truncate_after_first(match)
 
     for length in range(1, max_ngram_length):
         pattern = rf'(?<!\d)(\S{{{length}}})\1{{{max_occurrences - 1},}}(?!\d)'
         combined = rf'(?=.*\D){pattern}'
         match = regex.search(combined, cleaned)
         if match:
-            repeated = match.group(1)
-            pos = text.find(repeated)
-            if pos >= 0:
-                end_pos = text.find(repeated, pos + len(repeated))
-                if end_pos >= 0:
-                    return text[:end_pos + len(repeated)], True
-            return text[:len(text)//2], True
+            return truncate_after_first(match)
 
     return text, False
 
@@ -401,6 +475,12 @@ class RealtimeASRSession:
         self.prev_text = ""
         self.last_partial_text = ""
         self.last_partial_start_ms = 0
+        self.last_partial_end_ms = 0
+        self.last_partial_eligible = False
+        self.segment_partial_text = ""
+        self.segment_partial_start_ms = 0
+        self.segment_partial_end_ms = 0
+        self.segment_partial_stable_count = 0
         self.last_decode_samples = 0
         self.locked_sentences = []
         self.prev_seg_text = ""
@@ -428,15 +508,9 @@ class RealtimeASRSession:
             for seg in new_confirmed:
                 seg_text = self._decode_segment(seg)
                 self.prev_text = ""
-                if not seg_text.strip():
-                    continue
-                self.locked_sentences.append({"text": seg_text, "start": int(seg[0]), "end": int(seg[1])})
-                if self.spk_tracker:
-                    s0 = int(seg[0] * self.sample_rate / 1000)
-                    s1 = min(int(seg[1] * self.sample_rate / 1000), self.total_samples)
-                    segment_audio = self._slice_audio(s0, s1).copy()
-                    self.spk_tracker.assign_streaming(segment_audio, seg[0]/1000, seg[1]/1000, self.locked_sentences[-1])
-                logger.info(f"Locked: [{seg[0]}-{seg[1]}ms] \"{seg_text[:40]}\"")
+                self._append_completed_segment(seg, seg_text)
+                if seg_text.strip():
+                    logger.info(f"Locked: [{seg[0]}-{seg[1]}ms] \"{seg_text[:40]}\"")
 
         self._compact_audio_buffer()
 
@@ -464,6 +538,67 @@ class RealtimeASRSession:
         self.audio_buffer = np.array([], dtype=np.float32)
         self.audio_buffer_start_sample = self.total_samples
 
+    def _reset_partial_history(self):
+        self.segment_partial_text = ""
+        self.segment_partial_start_ms = 0
+        self.segment_partial_end_ms = 0
+        self.segment_partial_stable_count = 0
+
+    def _record_partial_text(self, text, start_ms, hallucinated=False):
+        """Build a confidence-bearing transcript across overlapping partial windows."""
+        end_ms = int(self.total_samples * 1000 / self.sample_rate)
+        self.last_partial_end_ms = end_ms
+        self.last_partial_eligible = bool(text.strip()) and not hallucinated
+        if not text.strip() or hallucinated:
+            self._reset_partial_history()
+            return
+
+        start_ms = int(start_ms)
+        incoming_norm = _normalize_transcript(text)
+        if not self.segment_partial_text:
+            self.segment_partial_text = text
+            self.segment_partial_start_ms = start_ms
+            self.segment_partial_end_ms = end_ms
+            self.segment_partial_stable_count = 1
+            return
+
+        history_norm = _normalize_transcript(self.segment_partial_text)
+        if start_ms == self.segment_partial_start_ms:
+            if incoming_norm == history_norm:
+                self.segment_partial_text = text
+                self.segment_partial_end_ms = end_ms
+                self.segment_partial_stable_count += 1
+            else:
+                self.segment_partial_text = text
+                self.segment_partial_end_ms = end_ms
+                self.segment_partial_stable_count = 1
+            return
+
+        if start_ms < self.segment_partial_start_ms:
+            self.segment_partial_text = text
+            self.segment_partial_start_ms = start_ms
+            self.segment_partial_end_ms = end_ms
+            self.segment_partial_stable_count = 1
+            return
+
+        if start_ms > self.segment_partial_end_ms:
+            self.segment_partial_text = text
+            self.segment_partial_start_ms = start_ms
+            self.segment_partial_end_ms = end_ms
+            self.segment_partial_stable_count = 1
+            return
+
+        merged = _merge_overlapping_transcripts(self.segment_partial_text, text)
+        if merged is not None:
+            self.segment_partial_text = merged
+            self.segment_partial_end_ms = end_ms
+            self.segment_partial_stable_count = 1
+        else:
+            self.segment_partial_text = text
+            self.segment_partial_start_ms = start_ms
+            self.segment_partial_end_ms = end_ms
+            self.segment_partial_stable_count = 1
+
     def should_decode(self):
         threshold = self.first_chunk_samples if not self.first_decode_done else self.chunk_samples
         return (self.total_samples - self.last_decode_samples) >= threshold
@@ -473,10 +608,11 @@ class RealtimeASRSession:
         if is_final and self.endpoint_mode == "client":
             return self.commit()
 
-        if self.endpoint_mode == "server" and self.total_samples < self.chunk_samples:
+        if self.endpoint_mode == "server":
             if is_final:
-                self._release_audio_buffer()
-            return self._build_response(is_final)
+                return self._finalize_server_session()
+            if self.total_samples < self.first_chunk_samples:
+                return self._build_response(is_final)
 
         if self.endpoint_mode == "client":
             if self.vad.current_speech_start is None:
@@ -487,25 +623,13 @@ class RealtimeASRSession:
             if self.total_samples - utterance_start_sample < self.first_chunk_samples:
                 return self._build_response(is_final=False)
 
-        if is_final:
-            if self.vad.current_speech_start is not None:
-                end_ms = int(self.total_samples * 1000 / self.sample_rate)
-                seg = [self.vad.current_speech_start, end_ms]
-                seg_text = self._decode_segment(seg)
-                if seg_text.strip():
-                    self.locked_sentences.append({"text": seg_text, "start": int(seg[0]), "end": int(seg[1])})
-                self.vad.current_speech_start = None
-
-            if self.spk_tracker and self.locked_sentences:
-                self.locked_sentences = self.spk_tracker.finalize(self.locked_sentences)
-            self._release_audio_buffer()
-            return self._build_response(is_final)
-
         if self.vad.current_speech_start is not None:
             seg_audio, partial_start_ms = self.get_partial_decode_audio()
         else:
             self.last_decode_samples = self.total_samples
             self.last_partial_text = ""
+            self.last_partial_eligible = False
+            self._reset_partial_history()
             return self._build_response(is_final)
 
         if len(seg_audio) < self.chunk_samples // 2:
@@ -523,6 +647,8 @@ class RealtimeASRSession:
             text = _clean_asr_text(text)
         except Exception as e:
             logger.error(f"ASR error: {e}")
+            self.last_partial_eligible = False
+            self._reset_partial_history()
             return self._build_response(is_final)
 
         text, hallucinated = detect_and_fix_hallucination(text)
@@ -532,6 +658,7 @@ class RealtimeASRSession:
         self.last_decode_samples = self.total_samples
         self.last_partial_text = text
         self.last_partial_start_ms = partial_start_ms
+        self._record_partial_text(text, partial_start_ms, hallucinated=hallucinated)
         if text.strip() and not self.first_decode_done:
             self.first_decode_done = True
 
@@ -554,31 +681,46 @@ class RealtimeASRSession:
             end_ms = int(self.total_samples * 1000 / self.sample_rate)
             seg = [self.vad.current_speech_start, end_ms]
             seg_text = self._decode_segment(seg)
-            if seg_text.strip():
-                sentence = {
-                    "text": seg_text,
-                    "start": int(seg[0]),
-                    "end": int(seg[1]),
-                }
-                if self.spk_tracker:
-                    s0 = int(seg[0] * self.sample_rate / 1000)
-                    segment_audio = self._slice_audio(s0, self.total_samples).copy()
-                    self.spk_tracker.assign_streaming(
-                        segment_audio,
-                        seg[0] / 1000,
-                        seg[1] / 1000,
-                        sentence,
-                    )
-                self.locked_sentences.append(sentence)
+            self._append_completed_segment(seg, seg_text)
             self.vad.current_speech_start = None
 
-        if self.spk_tracker and self.locked_sentences:
-            self.locked_sentences = self.spk_tracker.finalize(self.locked_sentences)
+        self._finalize_speakers()
 
         response = self._build_response(is_final=True)
         self._reset_utterance_state()
         return response
 
+    def _finalize_server_session(self):
+        duration_ms = int(self.total_samples * 1000 / self.sample_rate)
+        final_segments = []
+        finalize_vad = getattr(self.vad, "finalize", None)
+        if callable(finalize_vad):
+            try:
+                final_segments = finalize_vad() or []
+            except Exception as error:
+                logger.error(f"VAD finalization error: {error}")
+
+        had_valid_final_segment = False
+        for raw_seg in final_segments:
+            start_ms = max(0, int(raw_seg[0]))
+            end_ms = min(duration_ms, int(raw_seg[1]))
+            if end_ms <= start_ms:
+                continue
+            had_valid_final_segment = True
+            seg = [start_ms, end_ms]
+            seg_text = self._decode_segment(seg)
+            self._append_completed_segment(seg, seg_text)
+
+        if not had_valid_final_segment and self.vad.current_speech_start is not None:
+            seg = [int(self.vad.current_speech_start), duration_ms]
+            seg_text = self._decode_segment(seg)
+            self._append_completed_segment(seg, seg_text)
+
+        self.vad.current_speech_start = None
+        self._finalize_speakers()
+        self._release_audio_buffer()
+        return self._build_response(is_final=True)
+
     def _reset_utterance_state(self):
         self._release_audio_buffer()
         self.first_decode_done = False
@@ -586,6 +728,9 @@ class RealtimeASRSession:
         self.prev_text = ""
         self.last_partial_text = ""
         self.last_partial_start_ms = 0
+        self.last_partial_end_ms = 0
+        self.last_partial_eligible = False
+        self._reset_partial_history()
         self.last_decode_samples = self.total_samples
         self.locked_sentences = []
         self.prev_seg_text = ""
@@ -604,30 +749,140 @@ class RealtimeASRSession:
         start_ms = int(decode_start_sample * 1000 / self.sample_rate)
         return self._slice_audio(decode_start_sample, self.total_samples), start_ms
 
+    def _partial_fallback_candidate(self, seg, require_stable, latest_only=False):
+        if latest_only:
+            if not self.last_partial_eligible:
+                return ""
+            text = self.last_partial_text.strip()
+            start_ms = self.last_partial_start_ms
+            end_ms = self.last_partial_end_ms
+        else:
+            if require_stable and self.segment_partial_stable_count < 2:
+                return ""
+            text = self.segment_partial_text
+            start_ms = self.segment_partial_start_ms
+            end_ms = self.segment_partial_end_ms
+
+        end_tolerance_ms = 100
+        if (
+            not text
+            or int(start_ms) != int(seg[0])
+            or abs(int(end_ms) - int(seg[1])) > end_tolerance_ms
+        ):
+            return ""
+        return text
+
+    def _reconcile_completed_segment_text(self, decoded_text, seg, decode_succeeded):
+        """Keep a proven segment partial only when the completed decode regresses."""
+        final_text, final_hallucinated = detect_and_fix_hallucination(decoded_text)
+        if not decode_succeeded:
+            partial_text = self._partial_fallback_candidate(
+                seg, require_stable=False, latest_only=True
+            )
+            partial_text, _ = detect_and_fix_hallucination(partial_text)
+            if partial_text:
+                logger.warning(
+                    "Completed segment [%d-%dms] decode failed; keeping the latest "
+                    "full-coverage partial (%d chars)",
+                    int(seg[0]), int(seg[1]), len(partial_text),
+                )
+                return partial_text
+            return final_text
+
+        if not final_text.strip():
+            return final_text
+
+        partial_text = self._partial_fallback_candidate(
+            seg, require_stable=not final_hallucinated
+        )
+        if not partial_text:
+            return final_text
+        partial_text, partial_hallucinated = detect_and_fix_hallucination(partial_text)
+        final_cmp = _normalize_transcript(final_text)
+        partial_cmp = _normalize_transcript(partial_text)
+        segment_duration_ms = max(0, int(seg[1]) - int(seg[0]))
+        truncated_prefix = (
+            segment_duration_ms >= 2000
+            and len(final_cmp) >= 4
+            and partial_cmp.startswith(final_cmp)
+            and len(partial_cmp) >= len(final_cmp) * 2
+            and len(partial_cmp) - len(final_cmp) >= 8
+        )
+        use_partial = (
+            (final_hallucinated and not partial_hallucinated)
+            or truncated_prefix
+        )
+        if use_partial and partial_cmp:
+            reason = "hallucinated" if final_hallucinated else "truncated"
+            logger.warning(
+                "Completed segment [%d-%dms] decode was %s; "
+                "keeping the latest aligned partial (%d -> %d chars)",
+                int(seg[0]), int(seg[1]), reason, len(final_text), len(partial_text),
+            )
+            return partial_text
+        return final_text
+
+    def _clear_completed_partial(self, seg):
+        self.last_partial_text = ""
+        self.last_partial_start_ms = int(seg[1])
+        self.last_partial_end_ms = int(seg[1])
+        self.last_partial_eligible = False
+        self._reset_partial_history()
+
+    def _append_completed_segment(self, seg, text):
+        if not text.strip():
+            return
+        sentence = {
+            "text": text,
+            "start": int(seg[0]),
+            "end": int(seg[1]),
+        }
+        if self.spk_tracker:
+            s0 = int(seg[0] * self.sample_rate / 1000)
+            s1 = min(int(seg[1] * self.sample_rate / 1000), self.total_samples)
+            segment_audio = self._slice_audio(s0, s1).copy()
+            self.spk_tracker.assign_streaming(
+                segment_audio,
+                seg[0] / 1000,
+                seg[1] / 1000,
+                sentence,
+            )
+        self.locked_sentences.append(sentence)
+
+    def _finalize_speakers(self):
+        if self.spk_tracker and self.locked_sentences:
+            self.locked_sentences = self.spk_tracker.finalize(self.locked_sentences)
+
     @torch.no_grad()
     def _decode_segment(self, seg):
         """Decode a completed VAD segment via vLLM."""
         start_sample = int(seg[0] * self.sample_rate / 1000)
         end_sample = min(int(seg[1] * self.sample_rate / 1000), self.total_samples)
         seg_audio = self._slice_audio(start_sample, end_sample)
-        if len(seg_audio) < 1600:
-            return ""
-        audio_tensor = torch.from_numpy(seg_audio).float()
-        try:
-            results = self.vllm_engine.generate(
-                inputs=[audio_tensor],
-                hotwords=self.asr_kwargs.get("hotwords"),
-                language=self.asr_kwargs.get("language"),
-                max_new_tokens=512,
-            )
-            text = results[0]["text"] if results else ""
-            text = _clean_asr_text(text)
-            text = _postprocess_result_text(text, self.asr_kwargs)
-            self.prev_seg_text = text
-            return text
-        except Exception as e:
-            logger.error(f"Segment decode error: {e}")
-            return ""
+        text = ""
+        decode_succeeded = False
+        if len(seg_audio) >= 1600:
+            audio_tensor = torch.from_numpy(seg_audio).float()
+            try:
+                results = self.vllm_engine.generate(
+                    inputs=[audio_tensor],
+                    hotwords=self.asr_kwargs.get("hotwords"),
+                    language=self.asr_kwargs.get("language"),
+                    max_new_tokens=512,
+                )
+                text = results[0]["text"] if results else ""
+                text = _clean_asr_text(text)
+                decode_succeeded = True
+            except Exception as e:
+                logger.error(f"Segment decode error: {e}")
+
+        text = self._reconcile_completed_segment_text(
+            text, seg, decode_succeeded=decode_succeeded
+        )
+        text = _postprocess_result_text(text, self.asr_kwargs)
+        self.prev_seg_text = text
+        self._clear_completed_partial(seg)
+        return text
 
     def _build_response(self, is_final):
         duration_ms = int(self.total_samples * 1000 / self.sample_rate)
@@ -672,6 +927,9 @@ class RealtimeASRSession:
         self.prev_text = ""
         self.last_partial_text = ""
         self.last_partial_start_ms = 0
+        self.last_partial_end_ms = 0
+        self.last_partial_eligible = False
+        self._reset_partial_history()
         self.last_decode_samples = 0
         self.locked_sentences = []
         if self.spk_tracker:

tests/test_realtime_ws_service.py

@@ -258,6 +258,818 @@ class DummyEngine:
         return [{"text": "hello"}]
 
 
+class FixedTextEngine(DummyEngine):
+    def __init__(self, text):
+        super().__init__()
+        self.text = text
+
+    def generate(self, inputs, **kwargs):
+        self.input_lengths.extend(len(audio) for audio in inputs)
+        self.generate_kwargs.append(kwargs)
+        return [{"text": self.text}]
+
+
+class FailingTextEngine(DummyEngine):
+    def generate(self, inputs, **kwargs):
+        raise RuntimeError("synthetic final decode failure")
+
+
+class ScriptedTextEngine(DummyEngine):
+    def __init__(self, outputs):
+        super().__init__()
+        self.outputs = list(outputs)
+
+    def generate(self, inputs, **kwargs):
+        self.input_lengths.extend(len(audio) for audio in inputs)
+        self.generate_kwargs.append(kwargs)
+        output = self.outputs.pop(0)
+        if isinstance(output, Exception):
+            raise output
+        return [{"text": output}]
+
+
+class ControllableVad:
+    def __init__(self, sample_rate=16000):
+        self.sample_rate = sample_rate
+        self.current_speech_start = 0
+        self.total_samples = 0
+        self.lock_next = False
+
+    def feed(self, audio, is_final=False):
+        self.total_samples += len(audio)
+        if not self.lock_next:
+            return []
+        self.lock_next = False
+        self.current_speech_start = None
+        return [[0, int(self.total_samples * 1000 / self.sample_rate)]]
+
+    def start_utterance(self, start_ms):
+        if self.current_speech_start is None:
+            self.current_speech_start = start_ms
+
+    def reset(self):
+        self.current_speech_start = None
+
+
+class PreviousBoundaryVad(ControllableVad):
+    def feed(self, audio, is_final=False):
+        self.total_samples += len(audio)
+        if not self.lock_next:
+            return []
+        self.lock_next = False
+        self.current_speech_start = None
+        end_samples = self.total_samples - len(audio)
+        return [[0, int(end_samples * 1000 / self.sample_rate)]]
+
+
+class FinalFlushVad:
+    def __init__(self, sample_rate=16000):
+        self.sample_rate = sample_rate
+        self.current_speech_start = None
+        self.total_samples = 0
+        self.finalize_calls = 0
+
+    def feed(self, audio, is_final=False):
+        assert is_final is False
+        self.total_samples += len(audio)
+        return []
+
+    def finalize(self):
+        self.finalize_calls += 1
+        return [[0, int(self.total_samples * 1000 / self.sample_rate)]]
+
+    def reset(self):
+        self.current_speech_start = None
+
+
+class RecordingSpeakerTracker:
+    def __init__(self):
+        self.assigned = []
+        self.finalize_calls = 0
+
+    def assign_streaming(self, audio, start_sec, end_sec, sentence):
+        sentence["spk"] = 7
+        self.assigned.append((len(audio), start_sec, end_sec))
+
+    def finalize(self, sentences):
+        self.finalize_calls += 1
+        return sentences
+
+    def reset(self):
+        pass
+
+
+class OneShotSegmentVad:
+    def __init__(self, sample_rate=16000):
+        self.sample_rate = sample_rate
+        self.current_speech_start = 0
+        self.emitted = False
+
+    def feed(self, audio, is_final=False):
+        if self.emitted:
+            return []
+        self.emitted = True
+        self.current_speech_start = None
+        return [[0, int(len(audio) * 1000 / self.sample_rate)]]
+
+    def reset(self):
+        self.current_speech_start = None
+
+
+def make_one_shot_segment_session(
+    module, final_text, partial_text, partial_start_ms=0, engine=None
+):
+    sample_rate = 16000
+    session = module.RealtimeASRSession(
+        vllm_engine=engine or FixedTextEngine(final_text),
+        asr_kwargs={},
+        vad=OneShotSegmentVad(sample_rate),
+        sample_rate=sample_rate,
+        chunk_ms=960,
+    )
+    session.last_partial_text = partial_text
+    session.last_partial_start_ms = partial_start_ms
+    session.last_partial_end_ms = 2000
+    session.segment_partial_text = partial_text
+    session.segment_partial_start_ms = partial_start_ms
+    session.segment_partial_end_ms = 2000
+    session.segment_partial_stable_count = 2
+    session.last_partial_eligible = True
+    session.add_audio(np.zeros(2 * sample_rate, dtype=np.int16).tobytes())
+    return session
+
+
+def add_pcm(session, duration_ms, sample_rate=16000):
+    samples = int(duration_ms * sample_rate / 1000)
+    session.add_audio(np.zeros(samples, dtype=np.int16).tobytes())
+
+
+def decode_partial(session, duration_ms):
+    add_pcm(session, duration_ms, session.sample_rate)
+    assert session.should_decode()
+    return session.decode(is_final=False)
+
+
+def test_completed_segment_keeps_aligned_partial_when_final_is_truncated_prefix():
+    module = load_service_module()
+    partial = (
+        "当然可以。目前我们已经完成了第一阶段的调查研究工作，并收集了所有必要的数据。"
+        "接下来我们将进入分析阶段，预计在下周完成。"
+    )
+
+    session = make_one_shot_segment_session(module, "当然可以。", partial)
+
+    assert session.locked_sentences == [
+        {"text": partial, "start": 0, "end": 2000}
+    ]
+    assert session.last_partial_text == ""
+
+
+def test_completed_segment_keeps_clean_aligned_partial_when_final_hallucinates():
+    module = load_service_module()
+    partial = "不好意思，上午杭高架交通比较混乱一点，我们不晓得我们有点耽误。"
+    repeated = "好，拜拜。" * 80
+
+    session = make_one_shot_segment_session(module, repeated, partial)
+
+    assert session.locked_sentences == [
+        {"text": partial, "start": 0, "end": 2000}
+    ]
+    assert session.last_partial_text == ""
+
+
+def test_completed_segment_keeps_aligned_partial_when_final_decode_fails():
+    module = load_service_module()
+    partial = "最终解码异常时，已经稳定显示的内容不应整句丢失。"
+
+    session = make_one_shot_segment_session(
+        module,
+        "",
+        partial,
+        engine=FailingTextEngine(),
+    )
+
+    assert session.locked_sentences == [
+        {"text": partial, "start": 0, "end": 2000}
+    ]
+    assert session.last_partial_text == ""
+
+
+def test_completed_segment_keeps_distinct_final_correction_and_clears_partial():
+    module = load_service_module()
+
+    session = make_one_shot_segment_session(
+        module,
+        "感谢各位今天参加会议。",
+        "感谢各位今天参加会以。",
+    )
+
+    assert session.locked_sentences == [
+        {"text": "感谢各位今天参加会议。", "start": 0, "end": 2000}
+    ]
+    assert session.last_partial_text == ""
+
+
+def test_completed_segment_does_not_reuse_partial_from_another_window():
+    module = load_service_module()
+    stale_partial = "当然可以。目前我们已经完成了第一阶段的调查研究工作。"
+
+    session = make_one_shot_segment_session(
+        module,
+        "当然可以。",
+        stale_partial,
+        partial_start_ms=5000,
+    )
+
+    assert session.locked_sentences == [
+        {"text": "当然可以。", "start": 0, "end": 2000}
+    ]
+    assert session.last_partial_text == ""
+
+
+def test_hallucination_repair_handles_repeated_phrases_with_punctuation():
+    module = load_service_module()
+    prefix = "不好意思，上午杭高架交通比较混乱一点。"
+    text = prefix + "好，拜拜。" * 80
+
+    repaired, hallucinated = module.detect_and_fix_hallucination(text)
+
+    assert hallucinated is True
+    assert repaired == prefix + "好，拜拜。"
+
+
+def test_server_stop_under_one_chunk_keeps_partial_when_final_decode_fails():
+    module = load_service_module()
+    engine = ScriptedTextEngine(
+        ["已经稳定显示的短句。", RuntimeError("synthetic final decode failure")]
+    )
+    session = module.RealtimeASRSession(
+        engine,
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=960,
+    )
+
+    partial = decode_partial(session, 600)
+    assert partial["partial"] == "已经稳定显示的短句。"
+    assert len(engine.generate_kwargs) == 1
+    result = session.decode(is_final=True)
+
+    assert len(engine.generate_kwargs) == 2
+    assert engine.outputs == []
+    assert result["sentences"] == [
+        {"text": "已经稳定显示的短句。", "start": 0, "end": 600}
+    ]
+    assert session.last_partial_text == ""
+
+
+def test_successful_empty_final_does_not_restore_old_partial():
+    module = load_service_module()
+    partial = "会议取消。稍后另行通知。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, partial, ""]),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=500,
+    )
+
+    decode_partial(session, 500)
+    decode_partial(session, 500)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == []
+    assert session.last_partial_text == ""
+
+
+def test_short_dense_final_can_remove_a_speculative_partial_suffix():
+    module = load_service_module()
+    partial = "会议取消。稍后另行通知。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, partial, "会议取消。"]),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=500,
+    )
+
+    decode_partial(session, 500)
+    decode_partial(session, 500)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "会议取消。", "start": 0, "end": 1000}
+    ]
+
+
+def test_successful_final_can_remove_suffix_after_two_growing_partials():
+    module = load_service_module()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            [
+                "会议取消。稍后另行通知。",
+                "会议取消。稍后另行通知，请等待后续安排。",
+                "会议取消。",
+            ]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=1000,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "会议取消。", "start": 0, "end": 2000}
+    ]
+
+
+def test_regressed_partial_resets_confidence_before_final_correction():
+    module = load_service_module()
+    partial = "会议取消。稍后另行通知，请等待后续安排。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, partial, "会议取消。", "会议取消。"]),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=1000,
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "会议取消。", "start": 0, "end": 3000}
+    ]
+
+
+def test_vad_lock_keeps_stable_full_partial_and_assigns_speaker():
+    module = load_service_module()
+    partial = (
+        "当然可以。目前我们已经完成了第一阶段的调查研究工作，并收集了所有必要的数据。"
+        "接下来我们将进入分析阶段。"
+    )
+    vad = PreviousBoundaryVad()
+    speaker = RecordingSpeakerTracker()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, partial, "当然可以。"]),
+        {},
+        vad,
+        spk_tracker=speaker,
+        sample_rate=16000,
+        chunk_ms=1000,
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    vad.lock_next = True
+    add_pcm(session, 1000)
+
+    assert session.locked_sentences == [
+        {"text": partial, "start": 0, "end": 2000, "spk": 7}
+    ]
+    assert speaker.assigned == [(32000, 0.0, 2.0)]
+    assert speaker.finalize_calls == 0
+    assert session.last_partial_text == ""
+
+
+def test_client_commit_uses_the_same_stable_partial_reconciliation():
+    module = load_service_module()
+    partial = (
+        "当然可以。目前我们已经完成了第一阶段的调查研究工作，并收集了所有必要的数据。"
+        "接下来我们将进入分析阶段。"
+    )
+    speaker = RecordingSpeakerTracker()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, partial, "当然可以。"]),
+        {},
+        ControllableVad(),
+        spk_tracker=speaker,
+        sample_rate=16000,
+        chunk_ms=1000,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": partial, "start": 0, "end": 2000, "spk": 7}
+    ]
+    assert speaker.finalize_calls == 1
+    assert session.last_partial_text == ""
+
+
+def test_shifted_partial_without_full_segment_history_is_not_reused():
+    module = load_service_module()
+    partial = "当然可以。目前我们已经完成了全部调查研究工作。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, "当然可以。"]),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=960,
+        partial_window_sec=1.5,
+    )
+
+    decode_partial(session, 2000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "当然可以。", "start": 0, "end": 2000}
+    ]
+
+
+def test_partial_missing_the_segment_tail_is_not_reused():
+    module = load_service_module()
+    partial = "当然可以。目前我们已经完成了全部调查研究工作，并进入分析阶段。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, partial, "当然可以。"]),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=1000,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    add_pcm(session, 1000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "当然可以。", "start": 0, "end": 3000}
+    ]
+
+
+def test_nonoverlapping_partial_invalidates_old_full_segment_history():
+    module = load_service_module()
+    partial = "当然可以。目前我们已经完成了全部调查研究工作，并进入分析阶段。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            [
+                partial,
+                partial,
+                "完全无关的新窗口内容",
+                "仍然无关的后续窗口内容",
+                "当然可以。",
+            ]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=500,
+        partial_window_sec=1.0,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 500)
+    decode_partial(session, 500)
+    decode_partial(session, 500)
+    decode_partial(session, 500)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "当然可以。", "start": 0, "end": 2000}
+    ]
+
+
+def test_hallucinated_partial_is_not_locked_when_final_decode_fails():
+    module = load_service_module()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            ["好，拜拜。" * 80, RuntimeError("synthetic final decode failure")]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=500,
+        endpoint_mode="client",
+    )
+
+    partial = decode_partial(session, 500)
+    assert partial["partial"] == "好，拜拜。"
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == []
+
+
+def test_hallucinated_partial_breaks_stable_partial_continuity():
+    module = load_service_module()
+    partial = "会议取消。稍后另行通知，请等待后续安排。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            [partial, "好，拜拜。" * 80, partial, "会议取消。"]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=1000,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "会议取消。", "start": 0, "end": 3000}
+    ]
+
+
+def test_partial_decode_error_breaks_stable_partial_continuity():
+    module = load_service_module()
+    partial = "会议取消。稍后另行通知，请等待后续安排。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            [partial, RuntimeError("synthetic partial failure"), partial, "会议取消。"]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=1000,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "会议取消。", "start": 0, "end": 3000}
+    ]
+
+
+def test_empty_partial_breaks_stable_partial_continuity():
+    module = load_service_module()
+    partial = "会议取消。稍后另行通知，请等待后续安排。"
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine([partial, "", partial, "会议取消。"]),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=1000,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    decode_partial(session, 1000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "会议取消。", "start": 0, "end": 3000}
+    ]
+
+
+def test_partial_overlap_cannot_bridge_an_audio_coverage_gap():
+    module = load_service_module()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            ["ABCDEFGH", "EFGHIJKLMNOP", "好，拜拜。" * 80]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=500,
+        partial_window_sec=1.0,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 500)
+    decode_partial(session, 2000)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "好，拜拜。", "start": 0, "end": 2500}
+    ]
+
+
+def test_partial_overlap_cannot_bridge_a_100ms_audio_gap():
+    module = load_service_module()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            ["ABCDEFGH", "EFGHIJKLMNOP", "好，拜拜。" * 80]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=500,
+        partial_window_sec=0.5,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 500)
+    decode_partial(session, 600)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "好，拜拜。", "start": 0, "end": 1100}
+    ]
+
+
+def test_partial_must_start_exactly_at_segment_start_for_fallback():
+    module = load_service_module()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            ["ABCDEFGHIJKLMNOP", "ABCDEFGHIJKLMNOP", "ABCD"]
+        ),
+        {},
+        ControllableVad(),
+        sample_rate=16000,
+        chunk_ms=500,
+        partial_window_sec=1.9,
+        endpoint_mode="client",
+    )
+
+    decode_partial(session, 2000)
+    session.decode(is_final=False)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "ABCD", "start": 0, "end": 2000}
+    ]
+
+
+def test_server_stop_assigns_speaker_to_active_segment():
+    module = load_service_module()
+    speaker = RecordingSpeakerTracker()
+    session = module.RealtimeASRSession(
+        FixedTextEngine("已经完成的短句。"),
+        {},
+        ControllableVad(),
+        spk_tracker=speaker,
+        sample_rate=16000,
+        chunk_ms=960,
+    )
+
+    add_pcm(session, 600)
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {"text": "已经完成的短句。", "start": 0, "end": 600, "spk": 7}
+    ]
+    assert speaker.assigned == [(9600, 0.0, 0.6)]
+    assert speaker.finalize_calls == 1
+
+
+def test_server_stop_flushes_vad_before_decoding_final_segment():
+    module = load_service_module()
+    vad = FinalFlushVad()
+    speaker = RecordingSpeakerTracker()
+    engine = FixedTextEngine("由 VAD final flush 确认的短句。")
+    session = module.RealtimeASRSession(
+        engine,
+        {},
+        vad,
+        spk_tracker=speaker,
+        sample_rate=16000,
+        chunk_ms=960,
+    )
+
+    add_pcm(session, 600)
+    result = session.decode(is_final=True)
+
+    assert vad.finalize_calls == 1
+    assert len(engine.generate_kwargs) == 1
+    assert result["sentences"] == [
+        {
+            "text": "由 VAD final flush 确认的短句。",
+            "start": 0,
+            "end": 600,
+            "spk": 7,
+        }
+    ]
+    assert speaker.finalize_calls == 1
+
+
+def test_server_stop_finalizes_speakers_after_vad_already_locked():
+    module = load_service_module()
+    vad = ControllableVad()
+    speaker = RecordingSpeakerTracker()
+    session = module.RealtimeASRSession(
+        FixedTextEngine("已经由 VAD 锁定的句子。"),
+        {},
+        vad,
+        spk_tracker=speaker,
+        sample_rate=16000,
+        chunk_ms=960,
+    )
+
+    vad.lock_next = True
+    add_pcm(session, 600)
+    assert speaker.finalize_calls == 0
+    result = session.decode(is_final=True)
+
+    assert result["sentences"] == [
+        {
+            "text": "已经由 VAD 锁定的句子。",
+            "start": 0,
+            "end": 600,
+            "spk": 7,
+        }
+    ]
+    assert speaker.finalize_calls == 1
+
+
+def test_overlapping_partial_windows_preserve_long_segment_on_final_regression():
+    module = load_service_module()
+    vad = PreviousBoundaryVad()
+    session = module.RealtimeASRSession(
+        ScriptedTextEngine(
+            [
+                "ABCD",
+                "ABCDEFG",
+                "ABCDEFGHI",
+                "ABCDEFGHIJ",
+                "DEFGHIJKLMNO",
+                "好，拜拜。" * 80,
+            ]
+        ),
+        {},
+        vad,
+        sample_rate=16000,
+        chunk_ms=500,
+        partial_window_sec=2.0,
+    )
+
+    for _ in range(5):
+        decode_partial(session, 500)
+    vad.lock_next = True
+    add_pcm(session, 500)
+
+    assert session.locked_sentences == [
+        {"text": "ABCDEFGHIJKLMNO", "start": 0, "end": 2500}
+    ]
+    assert session.last_partial_text == ""
+
+
+def test_hallucination_repair_stops_before_the_next_opening_quote():
+    module = load_service_module()
+    prefix = "前缀："
+
+    repaired, hallucinated = module.detect_and_fix_hallucination(
+        prefix + "“好”" * 80
+    )
+
+    assert hallucinated is True
+    assert repaired == prefix + "“好”"
+
+
+def test_overlap_merge_preserves_casefold_expansion_source_positions():
+    module = load_service_module()
+
+    assert module._merge_overlapping_transcripts(
+        "Die Straße", "Straße ist frei"
+    ) == "Die Straße ist frei"
+
+
+def test_overlap_merge_preserves_new_punctuation_after_overlap():
+    module = load_service_module()
+
+    assert module._merge_overlapping_transcripts(
+        "今天天气", "今天天气，真好"
+    ) == "今天天气，真好"
+
+
+def test_hallucination_repair_keeps_one_ascii_quoted_phrase():
+    module = load_service_module()
+
+    repaired, hallucinated = module.detect_and_fix_hallucination(
+        "前缀：" + '"good"' * 80
+    )
+
+    assert hallucinated is True
+    assert repaired == '前缀："good"'
+
+
+def test_hallucination_repair_distinguishes_apostrophe_from_ascii_quote():
+    module = load_service_module()
+
+    repaired, hallucinated = module.detect_and_fix_hallucination(
+        "don't " + "'good'" * 80
+    )
+
+    assert hallucinated is True
+    assert repaired == "don't 'good'"
+
+
+def test_overlap_merge_rejects_casefold_split_inside_source_character():
+    module = load_service_module()
+
+    assert module._merge_overlapping_transcripts("abcs", "abcßXYZ") is None
+
+
 def test_client_endpoint_mode_skips_fsmn_vad_model(monkeypatch):
     module = load_service_module()
     auto_model_calls = []