feat(lifelog): 录音改为 block-based diarization-first 管线（攒段→整块分轮转写）

弃掉「录一句、转一句、实时聚类」——短句(0.3-30s) embedding 噪声大，实时滑动聚类把噪声平均进 anchor，是声纹串人的根因。改为：录音端(RecorderService)：VAD 只负责存语音 + 检测块边界，删掉实时 transcribe/ embed/clusterer.assign。块关闭条件：静音≥5min 或块时长≥10min(长会议保活)。 - BlockBuffer：speech-only PCM 追加 + manifest(run 偏移→真实时间)；崩溃重启 listPending 续处理。处理端(BlockProcessor)：整块 pyannote 分轮 → 每 turn 切片 paraformer 转写 + CAMPPlus 嵌入 + 全局 anchor 匹配 → 写 turn 行(turn = 时间轴一行)。pyannote 缺失则降级整块单 turn。块处理与采集解耦(Channel 顺序消费)，不阻塞录音。 embedding 从源头是单说话人干净切片，不再被实时逐句污染。SpeakerClusterer.assign 也加 L1 冻结(命名 anchor 不漂移)。废弃旧的周期 DiarizationWorker(块创建即 diarize)，删 DiarizationScheduler/Worker；纪要按钮不再触发它。实测(95s 真实会议块)：13 段语音 → 24 个干净不重叠 turn，逐 turn 带说话人+文字。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

feat(lifelog): 录音改为 block-based diarization-first 管线（攒段→整块分轮转写）
弃掉「录一句、转一句、实时聚类」——短句(0.3-30s) embedding 噪声大，实时滑动聚类把噪声平均进 anchor，是声纹串人的根因。改为：录音端(RecorderService)：VAD 只负责存语音 + 检测块边界，删掉实时 transcribe/ embed/clusterer.assign。块关闭条件：静音≥5min 或块时长≥10min(长会议保活)。 - BlockBuffer：speech-only PCM 追加 + manifest(run 偏移→真实时间)；崩溃重启 listPending 续处理。处理端(BlockProcessor)：整块 pyannote 分轮 → 每 turn 切片 paraformer 转写 + CAMPPlus 嵌入 + 全局 anchor 匹配 → 写 turn 行(turn = 时间轴一行)。pyannote 缺失则降级整块单 turn。块处理与采集解耦(Channel 顺序消费)，不阻塞录音。 embedding 从源头是单说话人干净切片，不再被实时逐句污染。SpeakerClusterer.assign 也加 L1 冻结(命名 anchor 不漂移)。废弃旧的周期 DiarizationWorker(块创建即 diarize)，删 DiarizationScheduler/Worker；纪要按钮不再触发它。实测(95s 真实会议块)：13 段语音 → 24 个干净不重叠 turn，逐 turn 带说话人+文字。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
yibo
1 parent c3220da9
Showing 8 changed files with 323 additions and 323 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/SpeakerClusterer.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
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockBuffer.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/BlockProcessor.kt
lifelog-recorder/app/src/main/kotlin/com/lemon/lifelog/service/RecorderService.kt
@@ -3,7 +3,6 @@ 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
@@ -18,9 +17,10 @@ class LifelogApp : Application() {
                 NotificationManager.IMPORTANCE_LOW,
             )
         )
-        // 纪要 + diarization periodic worker（各自 15min，UPDATE）。
+        // 纪要 periodic worker（15min，UPDATE）。
+        // 注：旧的周期 DiarizationWorker 已废弃——diarization-first 管线在「块」关闭时
+        // 当场分轮+转写，不再需要事后整体重跑。声纹纠偏走「重新聚类」按钮（reclusterAll）。
         SummaryScheduler.schedulePeriodic(this)
-        DiarizationScheduler.schedulePeriodic(this)
         // 相册：未处理照片的周期后台处理（人脸+VL），空闲时跑，不阻塞导入。
         com.lemon.lifelog.vision.PhotoScheduler.schedulePeriodic(this)
     }
@@ -83,7 +83,6 @@ 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
@@ -905,7 +904,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 = { DiarizationScheduler.runNow(activity); SummaryScheduler.runNow(activity) }, enabled = !running) {
+            Button(onClick = { SummaryScheduler.runNow(activity) }, enabled = !running) {
                 if (running) {
                     CircularProgressIndicator(modifier = Modifier.size(16.dp), strokeWidth = 2.dp)
                     Spacer(Modifier.width(6.dp))
@@ -59,7 +59,12 @@ class SpeakerClusterer(
         }
         if (best != null && bestSim >= matchThreshold) {
-            // capped running mean：上限保证质心保持自适应、不被早期样本钉死。
+            if (autoIndexOf(best.name) == null) {
+                // 人工命名 anchor：只归属、**冻结质心不漂移**（同 reclusterAll 的 L1）。
+                // 否则 anchor 吸入相邻说话人后会朝其漂移、再吞更多 → 雪球式串人。
+                return@withLock Assignment(best.id, bestSim, best.name)
+            }
+            // 自动簇：capped running mean，质心保持自适应、不被早期样本钉死。
             val n = minOf(best.count, maxSamplesPerCentroid)
             val merged = FloatArray(emb.size) { i -> (best.centroid[i] * n + emb[i]) / (n + 1) }
             l2normalize(merged)
-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.asr.toLeBytes
-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)
-        // WorkManager 给每个 worker 硬上限 10min，超时直接 kill。预算压到 5min，
-        // 留足余量让最后一块 diarize 收尾 + 全局 recluster + 写 cursor，绝不撞红线。
-        const val RUN_BUDGET_MS = 5 * 60 * 1000L
-        const val MIN_SESSION_SEGMENTS = 2          // 单段会话无需 diarize
-        // 单块 ≤3min speech：一次 diarize ~3min，远小于 10min 红线，一轮能跑完多块、
-        // 多次写 cursor 推进；块内说话人 index 不跨块——身份交全局 reclusterAll。
-        const val CHUNK_SAMPLES = 3 * 60 * 16000
-        private const val KEY_DONE = "diarized_sessions" // prefs StringSet of windowStartTs
-        private const val KEY_CURSOR = "diar_cursor_"    // + key → 已处理段数（断点续跑长会议）
-    }
-
-    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
-        var attempted = false
-        try {
-            // 最新会话优先（用户更关心最近的；旧会话后续轮次补）。
-            for (session in sessions.reversed()) {
-                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)
-                }
-
-                attempted = true
-                val startIdx = prefs.getInt(KEY_CURSOR + key, 0)
-                val reached = diarizeSession(repo, session, engine!!, embedder!!, runStart, startIdx)
-                if (reached >= session.size) {
-                    done.add(key); processed++
-                    prefs.edit().remove(KEY_CURSOR + key).apply()
-                } else {
-                    // 预算中途用完：记录断点，下轮从此处续跑（不重头），然后退出本轮。
-                    prefs.edit().putInt(KEY_CURSOR + key, reached).apply()
-                    Log.i(TAG, "session $key partial: cursor=$reached/${session.size}, resume next round"); break
-                }
-            }
-        } finally {
-            engine?.release()
-            embedder?.release()
-        }
-
-        // 全局重聚类：用 diarization 产出的"干净"per-turn embedding 统一聚类
-        //（reclusterAll 本就全局 + 保留人工命名锚点）。只要这轮碰过会话就跑。
-        if (attempted) {
-            val n = SpeakerClusterer.reclusterAll(repo, matchThreshold = threshold)
-            Log.i(TAG, "global recluster → $n speakers")
-        }
-        prefs.edit().putStringSet(KEY_DONE, done).apply()
-        Log.i(TAG, "done: ${sessions.size} sessions, $processed diarized")
-        return Result.success()
-    }
-
-    /**
-     * 分块 diarize 整个会话（长会议不再只取前 N min），从 [startIdx] 段断点续跑。
-     * 返回**已处理到的段下标**（== session.size 表示整场跑完；< 表示预算中途用完，
-     * 外层据此存断点、下轮从这里续跑——块内说话人 index 不跨块，身份交全局 reclusterAll）。
-     */
-    private suspend fun diarizeSession(
-        repo: Repository,
-        session: List<Segment>,
-        engine: DiarizationEngine,
-        embedder: SpeakerEmbedder,
-        runStart: Long,
-        startIdx: Int,
-    ): Int = withContext(Dispatchers.Default) {
-        var idx = startIdx.coerceIn(0, session.size)
-        var nChunks = 0
-        var totalTurns = 0
-        var embedded = 0
-        while (idx < session.size) {
-            if (System.currentTimeMillis() - runStart > RUN_BUDGET_MS) {
-                Log.i(TAG, "session ${session.first().startTs}: budget hit at seg $idx/${session.size}")
-                return@withContext idx  // 断点，外层续跑
-            }
-            // 攒一块（≤CHUNK_SAMPLES），解码即拼，记录每段在块内的 [from,to]
-            val segs = ArrayList<Segment>()
-            val froms = ArrayList<Int>()
-            val tos = ArrayList<Int>()
-            val pcms = ArrayList<FloatArray>()
-            var len = 0
-            while (idx < session.size && len < CHUNK_SAMPLES) {
-                val seg = session[idx]; idx++
-                val pcm = AudioDecoder.decodeToFloat(seg.audioPath)
-                if (pcm.isEmpty()) continue
-                segs.add(seg); froms.add(len); pcms.add(pcm); len += pcm.size; tos.add(len)
-            }
-            if (segs.isEmpty()) continue
-            val samples = FloatArray(len)
-            var off = 0
-            for (p in pcms) { p.copyInto(samples, off); off += p.size }
-            pcms.clear()
-
-            val turns = engine.diarize(samples)
-            if (turns.isEmpty()) continue
-            nChunks++; totalTurns += turns.size
-
-            val turnEmb = HashMap<Int, FloatArray?>()  // turn 下标 → embedding（多段共用同一 turn 缓存）
-            for (k in segs.indices) {
-                val s0 = froms[k].toFloat() / sr
-                val s1 = tos[k].toFloat() / sr
-                var bestIdx = -1
-                var bestOv = 0f
-                turns.forEachIndexed { i, t ->
-                    val ov = minOf(s1, t.endSec) - maxOf(s0, t.startSec)
-                    if (ov > bestOv) { bestOv = ov; bestIdx = i }
-                }
-                if (bestIdx < 0) continue
-                val emb = turnEmb.getOrPut(bestIdx) {
-                    val t = turns[bestIdx]
-                    val a = (t.startSec * sr).toInt().coerceIn(0, samples.size)
-                    val b = (t.endSec * sr).toInt().coerceIn(a, samples.size)
-                    if (b - a < sr / 4) null else embedder.embed(samples.copyOfRange(a, b), sr)
-                }
-                if (emb != null) {
-                    withContext(Dispatchers.IO) { repo.segments.updateEmbedding(segs[k].id, emb.toLeBytes()) }
-                    embedded++
-                }
-            }
-        }
-        Log.i(TAG, "session ${session.first().startTs}: $startIdx→${session.size} segs, $nChunks chunks, $totalTurns turns, $embedded clean-embedded")
-        idx  // == session.size，整场跑完
-    }
-}
+package com.lemon.lifelog.service
+
+import android.util.Log
+import java.io.File
+import java.nio.ByteBuffer
+import java.nio.ByteOrder
+import org.json.JSONArray
+import org.json.JSONObject
+
+/**
+ * 当前录音「块」的累积器（diarization-first 管线的录音端）。
+ *
+ * VAD 每产出一段语音就 [append] 进块的 `audio.pcm`（raw PCM16 LE，16k mono），并把该段的
+ * `{wallTs, sampleOffset, sampleLen}` 记进 manifest（用于之后把 turn 的样本偏移还原成真实时间）。
+ *
+ * 关块条件（[shouldClose]，由 RecorderService 每帧/每段检查）：
+ *   - 距上次语音静音 ≥ [SILENCE_MS]（5min）   或
+ *   - 块时长 ≥ [MAX_MS]（10min，长会议保活，避免全程不出文字）
+ *
+ * 块目录：`filesDir/blocks/<blockStartTs>/{audio.pcm, manifest.json}`。
+ * [close] 落 manifest(closed=true) 后交给 BlockProcessor；崩溃时块目录仍在，重启 [listPending] 重扫。
+ * 只存 speech（静音不落盘）→ 内存/磁盘按语音时长计，不按墙钟。
+ */
+class BlockBuffer(private val root: File, private val sampleRate: Int = 16000) {
+
+    data class Run(val ts: Long, val off: Int, val len: Int)
+
+    private var dir: File? = null
+    private var out: java.io.BufferedOutputStream? = null
+    private val runs = ArrayList<Run>()
+    private var sampleCount = 0
+    private var blockStartTs = 0L
+    private var lastSpeechEndTs = 0L
+
+    val isOpen get() = dir != null
+
+    init {
+        root.mkdirs()
+    }
+
+    /** 追加一段语音（startTs = 该段真实起始墙钟）。块未开则自动开。 */
+    fun append(samples: FloatArray, startTs: Long) {
+        if (samples.isEmpty()) return
+        if (!isOpen) open(startTs)
+        val bytes = ByteArray(samples.size * 2)
+        val bb = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN)
+        for (s in samples) bb.putShort((s.coerceIn(-1f, 1f) * 32767f).toInt().toShort())
+        out!!.write(bytes)
+        out!!.flush()
+        runs.add(Run(startTs, sampleCount, samples.size))
+        sampleCount += samples.size
+        lastSpeechEndTs = startTs + samples.size.toLong() * 1000 / sampleRate
+        writeManifest(dir!!, closed = false)
+    }
+
+    /** 在 [ts] 这一刻是否该关块（静音超时或块超长）。块未开恒 false。 */
+    fun shouldClose(ts: Long): Boolean {
+        if (!isOpen) return false
+        return (ts - lastSpeechEndTs >= SILENCE_MS) || (ts - blockStartTs >= MAX_MS)
+    }
+
+    /** 关块：落 manifest(closed=true)，返回块目录；空块直接删并返回 null。 */
+    fun close(): File? {
+        if (!isOpen) return null
+        runCatching { out?.flush(); out?.close() }
+        out = null
+        val d = dir!!
+        dir = null
+        if (sampleCount == 0) {
+            d.deleteRecursively()
+            return null
+        }
+        writeManifest(d, closed = true)
+        Log.i(TAG, "block close ${d.name}: ${runs.size} runs / $sampleCount samples")
+        return d
+    }
+
+    private fun open(ts: Long) {
+        blockStartTs = ts
+        lastSpeechEndTs = ts
+        runs.clear()
+        sampleCount = 0
+        val d = File(root, ts.toString()).apply { mkdirs() }
+        dir = d
+        out = File(d, "audio.pcm").outputStream().buffered()
+        Log.i(TAG, "block open @$ts -> ${d.name}")
+    }
+
+    private fun writeManifest(d: File, closed: Boolean) {
+        val arr = JSONArray()
+        for (r in runs) arr.put(JSONObject().put("ts", r.ts).put("off", r.off).put("len", r.len))
+        val obj = JSONObject()
+            .put("start", blockStartTs)
+            .put("sr", sampleRate)
+            .put("closed", closed)
+            .put("runs", arr)
+        runCatching { File(d, "manifest.json").writeText(obj.toString()) }
+    }
+
+    companion object {
+        private const val TAG = "BlockBuffer"
+        const val SILENCE_MS = 5 * 60 * 1000L   // 静音 ≥5min 切块
+        const val MAX_MS = 10 * 60 * 1000L      // 块 ≥10min 强切（长会议保活）
+
+        data class Loaded(val startTs: Long, val sampleRate: Int, val samples: FloatArray, val runs: List<Run>)
+
+        /** 启动时列出所有未处理块目录（崩溃恢复：处理后目录会被删除）。 */
+        fun listPending(root: File): List<File> =
+            root.listFiles()
+                ?.filter { it.isDirectory && File(it, "audio.pcm").exists() }
+                ?.sortedBy { it.name.toLongOrNull() ?: 0L }
+                ?: emptyList()
+
+        /** 读块：manifest + audio.pcm → 整块 float + runs。 */
+        fun load(dir: File): Loaded? {
+            val af = File(dir, "audio.pcm")
+            if (!af.exists()) return null
+            val mf = File(dir, "manifest.json")
+            var start = dir.name.toLongOrNull() ?: 0L
+            var sr = 16000
+            val runs = ArrayList<Run>()
+            if (mf.exists()) {
+                runCatching {
+                    val o = JSONObject(mf.readText())
+                    start = o.optLong("start", start)
+                    sr = o.optInt("sr", 16000)
+                    o.optJSONArray("runs")?.let { a ->
+                        for (i in 0 until a.length()) {
+                            val r = a.getJSONObject(i)
+                            runs.add(Run(r.getLong("ts"), r.getInt("off"), r.getInt("len")))
+                        }
+                    }
+                }
+            }
+            val bytes = af.readBytes()
+            val n = bytes.size / 2
+            val floats = FloatArray(n)
+            val bb = ByteBuffer.wrap(bytes).order(ByteOrder.LITTLE_ENDIAN)
+            for (i in 0 until n) floats[i] = bb.short / 32768f
+            // 崩溃时 manifest 可能没记全 runs → 合成一个整块 run（时间退化为连续）。
+            if (runs.isEmpty()) runs.add(Run(start, 0, n))
+            return Loaded(start, sr, floats, runs)
+        }
+    }
+}
+package com.lemon.lifelog.service
+
+import android.util.Log
+import com.lemon.lifelog.asr.DiarizationEngine
+import com.lemon.lifelog.asr.ParaformerEngine
+import com.lemon.lifelog.asr.SpeakerClusterer
+import com.lemon.lifelog.asr.SpeakerEmbedder
+import com.lemon.lifelog.asr.toLeBytes
+import com.lemon.lifelog.audio.SegmentWriter
+import com.lemon.lifelog.data.Repository
+import com.lemon.lifelog.data.Segment
+import com.lemon.lifelog.data.Transcript
+import java.io.File
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+
+/**
+ * 把一个关闭的录音块处理成时间轴 turns（diarization-first 管线的处理端）：
+ *
+ *   pyannote 分轮 → 每个 turn 切片 → paraformer 转写 + CAMPPlus 嵌入 + 全局 anchor 匹配
+ *   → 写 turn WAV + `segments` 行 + `transcripts` 行。
+ *
+ * 一个 **turn = 时间轴一行**（一人一段连续发言）。embedding 是单说话人的干净切片嵌入，
+ * 从源头避免了实时逐句聚类的噪声污染。处理完删块目录（= 已处理标记）。
+ *
+ * [diar] 为 null（pyannote 模型缺失）时降级：整块当一个 turn 转写。
+ */
+class BlockProcessor(
+    private val repo: Repository,
+    private val segWriter: SegmentWriter,
+    private val diar: DiarizationEngine?,
+    private val asr: ParaformerEngine,
+    private val embedder: SpeakerEmbedder?,
+    private val clusterer: SpeakerClusterer,
+) {
+    suspend fun process(dir: File) {
+        val loaded = withContext(Dispatchers.IO) { BlockBuffer.load(dir) }
+        if (loaded == null || loaded.samples.isEmpty()) {
+            withContext(Dispatchers.IO) { dir.deleteRecursively() }
+            return
+        }
+        val sr = loaded.sampleRate
+        val samples = loaded.samples
+        val runs = loaded.runs
+
+        // 样本下标 → 真实墙钟（speech-only 拼接，用 run 偏移还原）。runs 按 off 升序。
+        fun wallAt(sample: Int): Long {
+            var r = runs.first()
+            for (run in runs) if (run.off <= sample) r = run else break
+            val within = (sample - r.off).coerceIn(0, r.len)
+            return r.ts + within.toLong() * 1000 / sr
+        }
+
+        var turns = diar?.diarize(samples) ?: emptyList()
+        if (turns.isEmpty()) {
+            // 降级 / 极短块：整块一个 turn。
+            turns = listOf(DiarizationEngine.Turn(0f, samples.size.toFloat() / sr, 0))
+        }
+
+        var written = 0
+        for (t in turns) {
+            val a = (t.startSec * sr).toInt().coerceIn(0, samples.size)
+            val b = (t.endSec * sr).toInt().coerceIn(a, samples.size)
+            if (b - a < sr / 5) continue // <200ms 跳过
+            val slice = samples.copyOfRange(a, b)
+            val startTs = wallAt(a)
+            val durMs = (b - a).toLong() * 1000 / sr
+
+            val text = try {
+                asr.transcribeFloat(slice, sr).text
+            } catch (e: Throwable) {
+                Log.w(TAG, "asr failed", e); ""
+            }
+            val emb = embedder?.let {
+                try { it.embed(slice, sr) } catch (e: Throwable) { Log.w(TAG, "embed failed", e); null }
+            }
+            val assign = emb?.let { clusterer.assign(it, durMs) }
+                ?: SpeakerClusterer.Assignment(null, null, null)
+
+            val audioPath = withContext(Dispatchers.IO) { segWriter.write(slice, startTs, sr) }
+            withContext(Dispatchers.IO) {
+                val id = repo.segments.insert(
+                    Segment(
+                        startTs = startTs,
+                        endTs = startTs + durMs,
+                        audioPath = audioPath,
+                        durationMs = durMs,
+                        energyDb = 0f,
+                        embedding = emb?.toLeBytes(),
+                        speakerId = assign.speakerId,
+                        speakerSim = assign.sim,
+                    ),
+                )
+                if (text.isNotBlank()) repo.transcripts.insert(Transcript(segmentId = id, text = text))
+            }
+            written++
+        }
+        Log.i(TAG, "block ${dir.name}: ${turns.size} turns -> $written rows")
+        withContext(Dispatchers.IO) { dir.deleteRecursively() }
+    }
+
+    companion object {
+        private const val TAG = "BlockProcessor"
+    }
+}
@@ -14,23 +14,21 @@ import com.k2fsa.sherpa.onnx.SileroVadModelConfig
 import com.k2fsa.sherpa.onnx.Vad
 import com.k2fsa.sherpa.onnx.VadModelConfig
 import com.lemon.lifelog.MainActivity
+import com.lemon.lifelog.asr.DiarizationEngine
 import com.lemon.lifelog.asr.ModelDownloader
 import com.lemon.lifelog.asr.ParaformerEngine
 import com.lemon.lifelog.asr.SpeakerClusterer
 import com.lemon.lifelog.asr.SpeakerEmbedder
-import com.lemon.lifelog.asr.toLeBytes
 import com.lemon.lifelog.audio.AudioCapture
 import com.lemon.lifelog.audio.SegmentWriter
 import com.lemon.lifelog.data.Repository
-import com.lemon.lifelog.data.Segment
-import com.lemon.lifelog.data.SpeakerProfile
-import com.lemon.lifelog.data.Transcript
 import java.io.File
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.launch
@@ -70,10 +68,17 @@ class RecorderService : Service() {
     private var vad: Vad? = null
     private var engine: ParaformerEngine? = null
     private var speakerEmbedder: SpeakerEmbedder? = null
+    private var diarEngine: DiarizationEngine? = null
     /** 在线说话人聚类器（方案 B）：自动发现说话人并持久化为 speaker_profiles。 */
     private var clusterer: SpeakerClusterer? = null
+    /** diarization-first 管线：VAD 段攒进当前块，块关闭后整块 diarize+转写。 */
+    private lateinit var blockBuffer: BlockBuffer
+
+    /** 关闭的块排队顺序处理（与采集解耦，不阻塞录音）。 */
+    private val pendingBlocks = Channel<File>(Channel.UNLIMITED)
+
     override fun onCreate() {
         super.onCreate()
         startForegroundCompat()
@@ -83,6 +88,7 @@ class RecorderService : Service() {
         RecorderState.reset()
         repo = Repository(this)
         segWriter = SegmentWriter(File(filesDir, "segments"))
+        blockBuffer = BlockBuffer(File(filesDir, "blocks"))
         capture = AudioCapture()
         pipelineJob = scope.launch { runPipeline() }
@@ -113,7 +119,7 @@ class RecorderService : Service() {
         acquireWakeLock()
         capture.start(scope)
         RecorderState.setPaused(false)
-        RecorderState.setStatus("listening")
+        RecorderState.setStatus("录音中")
         notify("录音中")
         Log.i(TAG, "resumed")
     }
@@ -126,9 +132,13 @@ class RecorderService : Service() {
     override fun onDestroy() {
         pipelineJob?.cancel()
         capture.stop()
+        // 关掉当前块（落 manifest 留在磁盘，下次启动 listPending 续处理）。
+        runCatching { blockBuffer.close() }
+        pendingBlocks.close()
         vad?.release()
         engine?.release()
         speakerEmbedder?.release()
+        diarEngine?.release()
         scope.cancel()
         wakeLock?.takeIf { it.isHeld }?.release()
         wakeLock = null
@@ -210,94 +220,73 @@ class RecorderService : Service() {
             repo.speakers.observeAll().collect { clusterer?.load() }
         }
+        // 2e. Diarization 引擎（pyannote 分割 + CAMPPlus），块级整段分轮用。缺失则降级单 turn。
+        try {
+            RecorderState.setStatus("loading diarization…")
+            val pyannote = downloader.ensurePyannoteModel { Log.i(TAG, "pyannote dl: $it") }
+            val speaker = downloader.speakerPath()
+            if (pyannote.present && speaker.present) {
+                diarEngine = DiarizationEngine.load(
+                    pyannote.path, speaker.path, numThreads = 4,
+                    threshold = SpeakerClusterer.thresholdFromPrefs(this),
+                )
+            } else {
+                Log.w(TAG, "pyannote/speaker model missing; block will fall back to single-turn")
+            }
+        } catch (t: Throwable) {
+            Log.w(TAG, "diarization engine unavailable; single-turn fallback", t)
+        }
+
+        // 2f. 块处理器 + 顺序处理协程（与采集解耦）。
+        val processor = BlockProcessor(repo, segWriter, diarEngine, engine!!, speakerEmbedder, clusterer!!)
+        scope.launch {
+            for (dir in pendingBlocks) {
+                RecorderState.setStatus("处理段落…")
+                try {
+                    processor.process(dir)
+                } catch (t: Throwable) {
+                    Log.w(TAG, "block ${dir.name} process failed", t)
+                }
+                RecorderState.setStatus(if (paused) "已暂停" else "录音中")
+            }
+        }
+        // 崩溃恢复：把上次没处理完的块重新入队。
+        BlockBuffer.listPending(File(filesDir, "blocks")).forEach { pendingBlocks.trySend(it) }
+
         // 3. Start capturing (unless paused arrived during model load).
         if (!paused) {
             capture.start(scope)
-            RecorderState.setStatus("listening")
+            RecorderState.setStatus("录音中")
         }
-        Log.i(TAG, "pipeline running, clusterer loaded")
+        Log.i(TAG, "pipeline running (block-based diarization-first)")
         capture.frames.collect { samples ->
             val v = vad ?: return@collect
             v.acceptWaveform(samples)
+            val now = System.currentTimeMillis()
             while (!v.empty()) {
                 val seg = v.front()
                 v.pop()
                 if (seg.samples.isEmpty()) continue
-                handleSegment(seg.samples)
+                appendSpeech(seg.samples)
             }
+            // 静音超时 / 块超长 → 关块入队（即使本帧没有新语音也要检查）。
+            if (blockBuffer.shouldClose(now)) closeBlock()
         }
     }
-    private suspend fun handleSegment(samples: FloatArray) {
+    /** VAD 段 → 攒进当前块（不再实时转写/聚类）。跨静音/超长时先关旧块。 */
+    private fun appendSpeech(samples: FloatArray) {
         val sr = AudioCapture.SAMPLE_RATE
         val audioMs = samples.size.toLong() * 1000 / sr
-        val now = System.currentTimeMillis()
-        val startTs = now - audioMs
-
-        // a. persist audio (off CPU thread, Opus encode).
-        val audioPath = kotlinx.coroutines.withContext(Dispatchers.IO) {
-            segWriter.write(samples, startTs, sr)
-        }
-
-        // b. paraformer.
-        val wallStart = System.nanoTime()
-        val text = try {
-            engine!!.transcribeFloat(samples, sr).text
-        } catch (t: Throwable) {
-            Log.w(TAG, "transcribe failed", t)
-            ""
-        }
-        val wallMs = (System.nanoTime() - wallStart) / 1_000_000
-
-        // c. speaker embedding + online clustering (auto-discovers 说话人N).
-        val embedding: FloatArray? = speakerEmbedder?.let { embedder ->
-            try {
-                embedder.embed(samples, sr)
-            } catch (t: Throwable) {
-                Log.w(TAG, "embed failed", t)
-                null
-            }
-        }
-        val assignment = embedding?.let { emb ->
-            clusterer?.assign(emb, audioMs)
-        } ?: SpeakerClusterer.Assignment(null, null, null)
-        val matchedSpeakerId = assignment.speakerId
-        val matchedSim = assignment.sim
-        val matchedName = assignment.name
-
-        // d. Room insert.
-        val segmentId = kotlinx.coroutines.withContext(Dispatchers.IO) {
-            val id = repo.segments.insert(
-                Segment(
-                    startTs = startTs,
-                    endTs = now,
-                    audioPath = audioPath,
-                    durationMs = audioMs,
-                    energyDb = 0f,
-                    embedding = embedding?.toLeBytes(),
-                    speakerId = matchedSpeakerId,
-                    speakerSim = matchedSim,
-                ),
-            )
-            if (text.isNotEmpty()) {
-                repo.transcripts.insert(
-                    Transcript(
-                        segmentId = id,
-                        text = text,
-                    ),
-                )
-            }
-            id
-        }
+        val startTs = System.currentTimeMillis() - audioMs
+        if (blockBuffer.isOpen && blockBuffer.shouldClose(startTs)) closeBlock()
+        blockBuffer.append(samples, startTs)
+    }
-        Log.i(
-            TAG,
-            "segment#$segmentId audio=${audioMs}ms wall=${wallMs}ms " +
-                "speaker=${matchedName ?: "—"} sim=${matchedSim?.let { "%.2f".format(it) } ?: "—"} " +
-                "text=$text",
-        )
-        RecorderState.onSegment(text, audioMs, wallMs)
+    private fun closeBlock() {
+        val dir = blockBuffer.close() ?: return
+        pendingBlocks.trySend(dir)
     }
     private fun startForegroundCompat() {