feat(lifelog/diarize): 会话级 diarization 重跑（方案 A）—— 解决段内多说话人

逐 VAD 段单 embedding 的实时方案无法处理「一段几秒内多人说话」（混合 embedding=噪声） + 在线贪心聚类过度合并。改为会话级批处理重跑（与纪要 worker 同构）： - DiarizationEngine：sherpa-onnx OfflineSpeakerDiarization（pyannote 分割检测说话人变更点+重叠 → CAMPPlus embedding → FastClustering 自动聚类）。 - AudioDecoder：.opus(Opus-in-Ogg) → 16k mono PCM（MediaCodec，含 48k→16k 重采样 + 下混）。 - DiarizationWorker：每个已结束/未 diarize 会话 → 解码拼接 → diarize → 局部说话人桥接全局 SpeakerProfile（复用 SpeakerClusterer）→ 段按重叠最大 turn 重标。 AtomicBoolean 互斥 + 7min 预算 + 进程内 idempotency(prefs)。 - DiarizationScheduler：periodic 15min(UPDATE) + 「现在跑一次」一并触发。 - ModelDownloader.ensurePyannoteModel：首启下 pyannote-seg(~6MB)。实测端到端通：56 段会话 → 44 turns、检测出多说话人（核心能力达成）。已知待调（非 bug）： - 测试数据是有声书(一人配几十角色)，对声纹是地狱级输入 → 全局说话人偏多(98)。需在真实多人录音上验证；threshold 走设置页可调。 - 全局身份桥接现为贪心 match-or-create，易累积；v1.1 应改全局重聚类。 - 段内多说话人 v1 先归主导 turn，不拆分+重 ASR（v2）。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

feat(lifelog/diarize): 会话级 diarization 重跑（方案 A）—— 解决段内多说话人
逐 VAD 段单 embedding 的实时方案无法处理「一段几秒内多人说话」（混合 embedding=噪声） + 在线贪心聚类过度合并。改为会话级批处理重跑（与纪要 worker 同构）： - DiarizationEngine：sherpa-onnx OfflineSpeakerDiarization（pyannote 分割检测说话人变更点+重叠 → CAMPPlus embedding → FastClustering 自动聚类）。 - AudioDecoder：.opus(Opus-in-Ogg) → 16k mono PCM（MediaCodec，含 48k→16k 重采样 + 下混）。 - DiarizationWorker：每个已结束/未 diarize 会话 → 解码拼接 → diarize → 局部说话人桥接全局 SpeakerProfile（复用 SpeakerClusterer）→ 段按重叠最大 turn 重标。 AtomicBoolean 互斥 + 7min 预算 + 进程内 idempotency(prefs)。 - DiarizationScheduler：periodic 15min(UPDATE) + 「现在跑一次」一并触发。 - ModelDownloader.ensurePyannoteModel：首启下 pyannote-seg(~6MB)。实测端到端通：56 段会话 → 44 turns、检测出多说话人（核心能力达成）。已知待调（非 bug）： - 测试数据是有声书(一人配几十角色)，对声纹是地狱级输入 → 全局说话人偏多(98)。需在真实多人录音上验证；threshold 走设置页可调。 - 全局身份桥接现为贪心 match-or-create，易累积；v1.1 应改全局重聚类。 - 段内多说话人 v1 先归主导 turn，不拆分+重 ASR（v2）。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
yibo
1 parent 1edfc66e
Showing 7 changed files with 429 additions and 2 deletions
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/LifelogApp.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/MainActivity.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/DiarizationEngine.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/asr/ModelDownloader.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/audio/AudioDecoder.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationScheduler.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/diarize/DiarizationWorker.kt
@@ -3,6 +3,7 @@ package com.lemon.lifelog
 import android.app.Application
 import android.app.NotificationChannel
 import android.app.NotificationManager
+import com.lemon.lifelog.diarize.DiarizationScheduler
 import com.lemon.lifelog.service.RecorderService
 import com.lemon.lifelog.summary.SummaryScheduler
@@ -17,7 +18,8 @@ class LifelogApp : Application() {
                 NotificationManager.IMPORTANCE_LOW,
             )
         )
-        // Hourly 纪要 worker (KEEP = 已排过就不动)。
+        // 纪要 + diarization periodic worker（各自 15min，UPDATE）。
         SummaryScheduler.schedulePeriodic(this)
+        DiarizationScheduler.schedulePeriodic(this)
     }
 }
@@ -83,6 +83,7 @@ import com.lemon.lifelog.asr.XfyunIatClient
 import com.lemon.lifelog.asr.transcribeWithStats
 import com.lemon.lifelog.audio.SegmentPlayer
 import com.lemon.lifelog.chat.ChatConfig
+import com.lemon.lifelog.diarize.DiarizationScheduler
 import com.lemon.lifelog.summary.SummaryProgress
 import com.lemon.lifelog.summary.SummaryRunLog
 import com.lemon.lifelog.data.Repository
@@ -848,7 +849,7 @@ private fun SummariesTab(repo: Repository, activity: Context, onJumpToTime: (Lon
     Column(modifier = Modifier.fillMaxSize().padding(16.dp)) {
         Row(verticalAlignment = Alignment.CenterVertically, horizontalArrangement = Arrangement.spacedBy(8.dp)) {
             Text("纪要（${summaries.size}）", style = MaterialTheme.typography.titleMedium)
-            Button(onClick = { SummaryScheduler.runNow(activity) }, enabled = !running) {
+            Button(onClick = { DiarizationScheduler.runNow(activity); SummaryScheduler.runNow(activity) }, enabled = !running) {
                 if (running) {
                     CircularProgressIndicator(modifier = Modifier.size(16.dp), strokeWidth = 2.dp)
                     Spacer(Modifier.width(6.dp))
+package com.lemon.lifelog.asr
+
+import android.util.Log
+import com.k2fsa.sherpa.onnx.FastClusteringConfig
+import com.k2fsa.sherpa.onnx.OfflineSpeakerDiarization
+import com.k2fsa.sherpa.onnx.OfflineSpeakerDiarizationConfig
+import com.k2fsa.sherpa.onnx.OfflineSpeakerSegmentationModelConfig
+import com.k2fsa.sherpa.onnx.OfflineSpeakerSegmentationPyannoteModelConfig
+import com.k2fsa.sherpa.onnx.SpeakerEmbeddingExtractorConfig
+import java.io.File
+
+/**
+ * 离线说话人 diarization（sherpa-onnx）：pyannote 分割（检测说话人变更点 + 重叠语音）
+ * + CAMPPlus embedding + FastClustering 自动聚类。跑在一整段会话音频上，输出
+ * [start,end]→speaker turns，能把「几秒内多人」的区域拆成一人一段。
+ *
+ * 与逐 VAD 段单 embedding 的实时方案不同——这是会话级批处理重跑，准确得多。
+ */
+class DiarizationEngine private constructor(private val sd: OfflineSpeakerDiarization) {
+
+    data class Turn(val startSec: Float, val endSec: Float, val speaker: Int)
+
+    /** 16k mono PCM float（[-1,1]）→ 说话人 turns（按时间排序）。 */
+    fun diarize(samples: FloatArray): List<Turn> =
+        sd.process(samples).map { Turn(it.start, it.end, it.speaker) }.sortedBy { it.startSec }
+
+    fun sampleRate(): Int = sd.sampleRate()
+
+    fun release() = sd.release()
+
+    companion object {
+        private const val TAG = "DiarizationEngine"
+
+        /**
+         * @param threshold FastClustering 余弦阈值；越大分越细（同 SpeakerClusterer 取值习惯）。
+         */
+        fun load(
+            segModelPath: String,
+            embModelPath: String,
+            numThreads: Int = 4,
+            threshold: Float = 0.5f,
+        ): DiarizationEngine {
+            require(File(segModelPath).exists()) { "pyannote seg model 缺失: $segModelPath" }
+            require(File(embModelPath).exists()) { "CAMPPlus emb model 缺失: $embModelPath" }
+            val config = OfflineSpeakerDiarizationConfig(
+                segmentation = OfflineSpeakerSegmentationModelConfig(
+                    pyannote = OfflineSpeakerSegmentationPyannoteModelConfig(model = segModelPath),
+                    numThreads = numThreads,
+                ),
+                embedding = SpeakerEmbeddingExtractorConfig(model = embModelPath, numThreads = numThreads),
+                clustering = FastClusteringConfig(numClusters = -1, threshold = threshold),
+                minDurationOn = 0.2f,
+                minDurationOff = 0.5f,
+            )
+            val t0 = System.nanoTime()
+            val sd = OfflineSpeakerDiarization(config = config)
+            Log.i(TAG, "loaded in ${(System.nanoTime() - t0) / 1_000_000}ms sr=${sd.sampleRate()} thr=$threshold")
+            return DiarizationEngine(sd)
+        }
+    }
+}
@@ -99,6 +99,31 @@ class ModelDownloader(private val ctx: Context) {
             speakerPath()
         }
+    fun pyannotePath(): SpeakerModelFile {
+        val model = File(ctx.filesDir, "models/pyannote-seg.onnx")
+        return SpeakerModelFile(model.absolutePath, model.exists() && model.length() > 4_000_000L)
+    }
+
+    /** pyannote-segmentation-3.0（~6MB）—— diarization 的分割模型。 */
+    suspend fun ensurePyannoteModel(progress: (String) -> Unit): SpeakerModelFile =
+        withContext(Dispatchers.IO) {
+            val state = pyannotePath()
+            if (state.present) {
+                progress("pyannote model present, skip"); return@withContext state
+            }
+            File(ctx.filesDir, "models").mkdirs()
+            val task = DownloadTask(
+                File(state.path),
+                listOf(
+                    "https://hf-mirror.com/csukuangfj/sherpa-onnx-pyannote-segmentation-3-0/resolve/main/model.onnx",
+                    "https://huggingface.co/csukuangfj/sherpa-onnx-pyannote-segmentation-3-0/resolve/main/model.onnx",
+                ),
+                expectedSize = 5_992_913L,
+            )
+            downloadWithResume(http(), task, progress)
+            pyannotePath()
+        }
+
     private fun http() = OkHttpClient.Builder()
         .connectTimeout(15, TimeUnit.SECONDS)
         .readTimeout(120, TimeUnit.SECONDS)
+package com.lemon.lifelog.audio
+
+import android.media.MediaCodec
+import android.media.MediaExtractor
+import android.media.MediaFormat
+import android.util.Log
+import java.io.File
+import java.nio.ByteOrder
+
+/**
+ * 把 SegmentWriter 写的 .opus（Opus-in-Ogg）解码回 16k mono PCM float（[-1,1]）。
+ *
+ * 两个坑：
+ *  - MediaCodec 的 Opus 解码常输出 48kHz（Opus 原生率），与 diarization 期望的 16k 不符 → 重采样。
+ *  - 编码是 mono，但保险起见处理 stereo → 下混。
+ */
+object AudioDecoder {
+    private const val TAG = "AudioDecoder"
+    private const val TARGET_SR = 16000
+
+    fun decodeToFloat(path: String): FloatArray {
+        if (!File(path).exists()) return FloatArray(0)
+        val extractor = MediaExtractor()
+        return try {
+            extractor.setDataSource(path)
+            var track = -1
+            var inFormat: MediaFormat? = null
+            for (i in 0 until extractor.trackCount) {
+                val f = extractor.getTrackFormat(i)
+                if (f.getString(MediaFormat.KEY_MIME)?.startsWith("audio/") == true) {
+                    track = i; inFormat = f; break
+                }
+            }
+            if (track < 0 || inFormat == null) return FloatArray(0)
+            extractor.selectTrack(track)
+
+            val codec = MediaCodec.createDecoderByType(inFormat.getString(MediaFormat.KEY_MIME)!!)
+            codec.configure(inFormat, null, null, 0)
+            codec.start()
+
+            var srcRate = inFormat.getIntOr(MediaFormat.KEY_SAMPLE_RATE, TARGET_SR)
+            var channels = inFormat.getIntOr(MediaFormat.KEY_CHANNEL_COUNT, 1)
+
+            val shorts = ArrayList<Short>(TARGET_SR * 30)
+            val info = MediaCodec.BufferInfo()
+            var inEos = false
+            var outEos = false
+            val timeout = 10_000L
+
+            while (!outEos) {
+                if (!inEos) {
+                    val ii = codec.dequeueInputBuffer(timeout)
+                    if (ii >= 0) {
+                        val buf = codec.getInputBuffer(ii)!!
+                        val n = extractor.readSampleData(buf, 0)
+                        if (n < 0) {
+                            codec.queueInputBuffer(ii, 0, 0, 0, MediaCodec.BUFFER_FLAG_END_OF_STREAM)
+                            inEos = true
+                        } else {
+                            codec.queueInputBuffer(ii, 0, n, extractor.sampleTime, 0)
+                            extractor.advance()
+                        }
+                    }
+                }
+                val oi = codec.dequeueOutputBuffer(info, timeout)
+                when {
+                    oi == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED -> {
+                        val of = codec.outputFormat
+                        srcRate = of.getIntOr(MediaFormat.KEY_SAMPLE_RATE, srcRate)
+                        channels = of.getIntOr(MediaFormat.KEY_CHANNEL_COUNT, channels)
+                    }
+                    oi >= 0 -> {
+                        if (info.size > 0) {
+                            val ob = codec.getOutputBuffer(oi)!!
+                            ob.position(info.offset); ob.limit(info.offset + info.size)
+                            val sb = ob.order(ByteOrder.LITTLE_ENDIAN).asShortBuffer()
+                            while (sb.hasRemaining()) shorts.add(sb.get())
+                        }
+                        codec.releaseOutputBuffer(oi, false)
+                        if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) outEos = true
+                    }
+                }
+            }
+            codec.stop(); codec.release()
+
+            // 下混 stereo → mono
+            var mono = if (channels >= 2) {
+                FloatArray(shorts.size / channels) { i ->
+                    var s = 0f
+                    for (c in 0 until channels) s += shorts[i * channels + c] / 32768f
+                    s / channels
+                }
+            } else {
+                FloatArray(shorts.size) { shorts[it] / 32768f }
+            }
+            if (srcRate != TARGET_SR) mono = resample(mono, srcRate, TARGET_SR)
+            mono
+        } catch (t: Throwable) {
+            Log.w(TAG, "decode failed $path: ${t.message}")
+            FloatArray(0)
+        } finally {
+            extractor.release()
+        }
+    }
+
+    private fun MediaFormat.getIntOr(key: String, def: Int): Int =
+        if (containsKey(key)) getInteger(key) else def
+
+    /** 线性重采样到 dstRate。 */
+    private fun resample(input: FloatArray, srcRate: Int, dstRate: Int): FloatArray {
+        if (srcRate == dstRate || input.isEmpty()) return input
+        val ratio = dstRate.toDouble() / srcRate
+        val outLen = (input.size * ratio).toInt()
+        val out = FloatArray(outLen)
+        val last = input.size - 1
+        for (i in out.indices) {
+            val srcPos = i / ratio
+            val i0 = srcPos.toInt()
+            val frac = (srcPos - i0).toFloat()
+            val a = input[i0.coerceIn(0, last)]
+            val b = input[(i0 + 1).coerceIn(0, last)]
+            out[i] = a + (b - a) * frac
+        }
+        return out
+    }
+}
+package com.lemon.lifelog.diarize
+
+import android.content.Context
+import androidx.work.Constraints
+import androidx.work.ExistingPeriodicWorkPolicy
+import androidx.work.ExistingWorkPolicy
+import androidx.work.NetworkType
+import androidx.work.OneTimeWorkRequestBuilder
+import androidx.work.PeriodicWorkRequestBuilder
+import androidx.work.WorkManager
+import java.util.concurrent.TimeUnit
+
+object DiarizationScheduler {
+    private const val PERIODIC_TAG = "lifelog-diarize-periodic"
+    private const val ONE_SHOT_TAG = "lifelog-diarize-once"
+
+    fun schedulePeriodic(ctx: Context) {
+        val constraints = Constraints.Builder()
+            .setRequiredNetworkType(NetworkType.NOT_REQUIRED)
+            .setRequiresBatteryNotLow(true)
+            .build()
+        val request = PeriodicWorkRequestBuilder<DiarizationWorker>(15, TimeUnit.MINUTES)
+            .setConstraints(constraints)
+            .build()
+        WorkManager.getInstance(ctx).enqueueUniquePeriodicWork(
+            PERIODIC_TAG,
+            ExistingPeriodicWorkPolicy.UPDATE,
+            request,
+        )
+    }
+
+    /** 手动跑一次（KEEP：已在跑/排队则忽略）。 */
+    fun runNow(ctx: Context) {
+        val request = OneTimeWorkRequestBuilder<DiarizationWorker>().addTag(ONE_SHOT_TAG).build()
+        WorkManager.getInstance(ctx).enqueueUniqueWork(ONE_SHOT_TAG, ExistingWorkPolicy.KEEP, request)
+    }
+}
+package com.lemon.lifelog.diarize
+
+import android.content.Context
+import android.util.Log
+import androidx.work.CoroutineWorker
+import androidx.work.WorkerParameters
+import com.lemon.lifelog.asr.DiarizationEngine
+import com.lemon.lifelog.asr.ModelDownloader
+import com.lemon.lifelog.asr.SpeakerClusterer
+import com.lemon.lifelog.asr.SpeakerEmbedder
+import com.lemon.lifelog.audio.AudioDecoder
+import com.lemon.lifelog.data.Repository
+import com.lemon.lifelog.data.Segment
+import com.lemon.lifelog.summary.SummaryWorker
+import java.io.File
+import java.util.concurrent.atomic.AtomicBoolean
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+
+/**
+ * 会话级 diarization 重跑（方案 A）。对每个**已结束、未 diarize**的会话：
+ *   1. 解码 + 按序拼接该会话所有段音频（speech-only 16k mono）。
+ *   2. pyannote 分割 + CAMPPlus + FastClustering → 准确的 [start,end]→局部说话人 turns
+ *      （能把「几秒多人」的段拆开）。
+ *   3. 每个局部说话人 → 取其最长 turn 的 embedding → 经 SpeakerClusterer 桥接到
+ *      全局 SpeakerProfile（跨会话/跨天「说话人N」一致）。
+ *   4. 每个 VAD 段 → 找重叠最大的 turn → 重标 speakerId（段内多说话人先归主导者，v1）。
+ *
+ * 与逐段实时聚类相比准确得多；实时标签只作即时占位，这里批处理纠正。
+ */
+class DiarizationWorker(ctx: Context, params: WorkerParameters) : CoroutineWorker(ctx, params) {
+
+    companion object {
+        const val TAG = "DiarizationWorker"
+        private val isRunning = AtomicBoolean(false)
+        const val RUN_BUDGET_MS = 7 * 60 * 1000L
+        const val MIN_SESSION_SEGMENTS = 2          // 单段会话无需 diarize
+        const val MAX_CONCAT_SAMPLES = 18 * 60 * 16000  // ≤18min speech，防内存/超时
+        private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs
+    }
+
+    private val sr = 16000
+
+    override suspend fun doWork(): Result {
+        if (!isRunning.compareAndSet(false, true)) {
+            Log.i(TAG, "another diarization in progress, skip"); return Result.success()
+        }
+        return try {
+            run()
+        } catch (t: Throwable) {
+            Log.w(TAG, "diarization run failed", t); Result.success()
+        } finally {
+            isRunning.set(false)
+        }
+    }
+
+    private suspend fun run(): Result {
+        val runStart = System.currentTimeMillis()
+        val now = System.currentTimeMillis()
+        val prefs = applicationContext.getSharedPreferences(SummaryWorker.PREFS, Context.MODE_PRIVATE)
+        val gapMs = prefs.getInt(SummaryWorker.KEY_GAP_MIN, SummaryWorker.DEFAULT_GAP_MIN) * 60_000L
+        val threshold = SpeakerClusterer.thresholdFromPrefs(applicationContext)
+
+        val repo = Repository(applicationContext)
+        val segments = repo.segments.page(limit = 5000, offset = 0).sortedBy { it.startTs }
+        if (segments.size < MIN_SESSION_SEGMENTS) return Result.success()
+
+        val sessions = SummaryWorker.groupSessions(segments, gapMs)
+        val done = prefs.getStringSet(KEY_DONE, emptySet())!!.toMutableSet()
+
+        // 模型路径
+        val downloader = ModelDownloader(applicationContext)
+        val speakerFile = downloader.ensureSpeakerModel { }
+        if (!speakerFile.present) { Log.w(TAG, "speaker model missing"); return Result.success() }
+        val pyannote = downloader.ensurePyannoteModel { Log.i(TAG, "pyannote: $it") }
+        if (!pyannote.present) { Log.w(TAG, "pyannote seg model missing/download failed"); return Result.success() }
+
+        var engine: DiarizationEngine? = null
+        var embedder: SpeakerEmbedder? = null
+        var processed = 0
+        try {
+            for (session in sessions) {
+                val start = session.first().startTs
+                val end = session.last().endTs
+                val key = start.toString()
+
+                if (now - end <= gapMs) continue            // 未结束
+                if (session.size < MIN_SESSION_SEGMENTS) continue
+                if (done.contains(key)) continue            // 已 diarize
+                if (System.currentTimeMillis() - runStart > RUN_BUDGET_MS) {
+                    Log.i(TAG, "time budget hit, rest next round"); break
+                }
+
+                // 懒加载模型（有活才加载）
+                if (engine == null) {
+                    engine = DiarizationEngine.load(pyannote.path, speakerFile.path, numThreads = 4, threshold = threshold)
+                    embedder = SpeakerEmbedder.load(speakerFile.path, numThreads = 2)
+                }
+
+                val ok = diarizeSession(repo, session, engine!!, embedder!!, threshold)
+                if (ok) { done.add(key); processed++ }
+            }
+        } finally {
+            engine?.release()
+            embedder?.release()
+        }
+
+        prefs.edit().putStringSet(KEY_DONE, done).apply()
+        Log.i(TAG, "done: ${sessions.size} sessions, $processed diarized")
+        return Result.success()
+    }
+
+    /** 对单个会话做 diarization + 段重标。返回是否成功处理。 */
+    private suspend fun diarizeSession(
+        repo: Repository,
+        session: List<Segment>,
+        engine: DiarizationEngine,
+        embedder: SpeakerEmbedder,
+        threshold: Float,
+    ): Boolean = withContext(Dispatchers.Default) {
+        // 1. 解码 + 拼接，记录每段在拼接波形里的 [offset, offset+len]（采样）
+        data class SegSpan(val seg: Segment, val from: Int, val to: Int)
+        val spans = ArrayList<SegSpan>(session.size)
+        val audio = ArrayList<Float>(sr * 60)
+        for (seg in session) {
+            val pcm = AudioDecoder.decodeToFloat(seg.audioPath)
+            if (pcm.isEmpty()) continue
+            val from = audio.size
+            for (v in pcm) audio.add(v)
+            spans.add(SegSpan(seg, from, audio.size))
+            if (audio.size >= MAX_CONCAT_SAMPLES) break
+        }
+        if (spans.size < MIN_SESSION_SEGMENTS || audio.isEmpty()) return@withContext false
+        val samples = FloatArray(audio.size) { audio[it] }
+
+        // 2. diarize
+        val turns = engine.diarize(samples)
+        if (turns.isEmpty()) return@withContext false
+
+        // 3. 每个局部说话人 → 最长 turn 的 embedding → 桥接全局
+        val byLocal = turns.groupBy { it.speaker }
+        val clusterer = SpeakerClusterer(repo, matchThreshold = threshold).also { it.load() }
+        val localToGlobal = HashMap<Int, Pair<Long, Float>>() // local → (globalSpeakerId, sim)
+        for ((local, ts) in byLocal) {
+            // 局部说话人总时长 <2s 视为噪声，不桥接（避免一堆短 turn 各建一个全局说话人）。
+            val totalSec = ts.sumOf { (it.endSec - it.startSec).toDouble() }
+            if (totalSec < 2.0) continue
+            val longest = ts.maxByOrNull { it.endSec - it.startSec } ?: continue
+            val a = (longest.startSec * sr).toInt().coerceIn(0, samples.size)
+            val b = (longest.endSec * sr).toInt().coerceIn(a, samples.size)
+            if (b - a < sr / 2) continue  // <0.5s 太短，跳过该局部说话人
+            val emb = embedder.embed(samples.copyOfRange(a, b), sr) ?: continue
+            val durMs = ((longest.endSec - longest.startSec) * 1000).toLong()
+            val asg = clusterer.assign(emb, durMs)
+            if (asg.speakerId != null) localToGlobal[local] = asg.speakerId to (asg.sim ?: 1f)
+        }
+        if (localToGlobal.isEmpty()) return@withContext false
+
+        // 4. 每段 → 重叠最大的 turn → 全局 speakerId
+        for (span in spans) {
+            val s0 = span.from.toFloat() / sr
+            val s1 = span.to.toFloat() / sr
+            var bestTurn: DiarizationEngine.Turn? = null
+            var bestOv = 0f
+            for (t in turns) {
+                val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec)
+                if (ov > bestOv) { bestOv = ov; bestTurn = t }
+            }
+            val g = bestTurn?.let { localToGlobal[it.speaker] } ?: continue
+            withContext(Dispatchers.IO) { repo.segments.updateSpeaker(span.seg.id, g.first, g.second) }
+        }
+        Log.i(TAG, "session ${session.first().startTs}: ${spans.size} segs, ${turns.size} turns, ${byLocal.size} local speakers")
+        true
+    }
+}