Add AI batch reconciliation for accessibility bills

2026-07-25 18:41:09 +08:00
parent eb8909a192
commit 7cca34b331
25 changed files with 2771 additions and 230 deletions
@@ -1,112 +0,0 @@
-package com.nx.miaoji
-
-import android.content.Context
-import android.util.Log
-import org.json.JSONObject
-import java.io.ByteArrayOutputStream
-import java.net.HttpURLConnection
-import java.net.URL
-import java.time.Instant
-import java.util.UUID
-import java.util.concurrent.Executors
-
-object BackgroundAiRecognizer {
-    private val executor = Executors.newSingleThreadExecutor()
-
-    fun analyze(
-        context: Context,
-        packageName: String,
-        image: ByteArray,
-        flowSessionId: String? = null,
-    ) {
-        val appContext = context.applicationContext
-        executor.execute {
-            try {
-                val settings = RecognitionSettings.snapshot(appContext)
-                val token = RecognitionSettings.runtimeToken(appContext)
-                val baseUrl = settings.baseUrl?.trimEnd('/')
-                if (!settings.aiScreenshot || !settings.aiAllowed || !settings.hasAccount ||
-                    token.isNullOrBlank() || baseUrl.isNullOrBlank()
-                ) {
-                    return@execute
-                }
-                val response = upload("$baseUrl/api/parse/image?source=screenshot", token, image)
-                if (response.first == HttpURLConnection.HTTP_FORBIDDEN &&
-                    response.second.contains("AI_PERMISSION_DENIED")
-                ) {
-                    RecognitionSettings.disableRuntimeAi(appContext)
-                    return@execute
-                }
-                if (response.first !in 200..299) {
-                    Log.w(TAG, "Background AI parse failed status=${response.first}")
-                    return@execute
-                }
-                val root = JSONObject(response.second)
-                val items = root.optJSONArray("items") ?: return@execute
-                for (index in 0 until minOf(items.length(), 10)) {
-                    val item = items.optJSONObject(index) ?: continue
-                    val type = item.optString("type").lowercase()
-                    val amount = item.optDouble("amount", 0.0)
-                    if (type !in setOf("income", "expense") || amount <= 0) continue
-                    val occurredAt = runCatching {
-                        Instant.parse(item.optString("occurredAt")).toEpochMilli()
-                    }.getOrElse { System.currentTimeMillis() }
-                    val note = item.optString("note").takeIf { it.isNotBlank() }
-                    RecognitionCoordinator.get(appContext).submit(
-                        PaymentSignal(
-                            packageName = packageName,
-                            channel = "recognition_ai",
-                            amountCents = kotlin.math.round(amount * 100).toLong(),
-                            type = type,
-                            merchant = note,
-                            orderId = null,
-                            occurredAtEpochMs = occurredAt,
-                            knownTemplate = false,
-                            sourceEventId = "ai:" + (flowSessionId ?: UUID.randomUUID().toString()),
-                            sourceText = "AI 截图补全 · " + PaymentParser.appName(packageName),
-                            flowSessionId = flowSessionId,
-                            evidenceConfidence = "confirm",
-                        ),
-                    )
-                    break
-                }
-            } catch (error: Exception) {
-                Log.w(TAG, "Background AI parse unavailable", error)
-            } finally {
-                image.fill(0)
-            }
-        }
-    }
-
-    private fun upload(
-        endpoint: String,
-        token: String,
-        image: ByteArray,
-    ): Pair<Int, String> {
-        val boundary = "----Jizhi${UUID.randomUUID()}"
-        val connection = URL(endpoint).openConnection() as HttpURLConnection
-        connection.connectTimeout = 10_000
-        connection.readTimeout = 120_000
-        connection.requestMethod = "POST"
-        connection.doOutput = true
-        connection.setRequestProperty("Authorization", "Bearer $token")
-        connection.setRequestProperty("Content-Type", "multipart/form-data; boundary=$boundary")
-        connection.outputStream.use { output ->
-            output.write("--$boundary\r\n".toByteArray())
-            output.write(
-                "Content-Disposition: form-data; name=\"file\"; filename=\"recognition.png\"\r\n"
-                    .toByteArray(),
-            )
-            output.write("Content-Type: image/png\r\n\r\n".toByteArray())
-            output.write(image)
-            output.write("\r\n--$boundary--\r\n".toByteArray())
-        }
-        val code = connection.responseCode
-        val stream = if (code in 200..299) connection.inputStream else connection.errorStream
-        val body = stream?.bufferedReader(Charsets.UTF_8)?.use { it.readText() }.orEmpty()
-        connection.disconnect()
-        return code to body
-    }
-
-    private const val TAG = "JizhiRecognition"
-}
@@ -42,7 +42,8 @@ class MainActivity : FlutterActivity() {
        const val ACTION_SCREENSHOT_ERROR = "screenshot_error"
        const val ACTION_RECOGNITION_CONFIRM = "recognition_confirm"
        const val ACTION_RECOGNITION_UNDO = "recognition_undo"
-        const val ACTION_RECOGNITION_EDIT = "recognition_edit"
+        const val ACTION_RECOGNITION_EDIT = "recognition_edit"
+        const val ACTION_RECOGNITION_BATCH_REVIEW = "recognition_batch_review"
        const val EXTRA_SCREENSHOT_PATH = "screenshotPath"
        const val EXTRA_SCREENSHOT_ERROR = "screenshotError"
        const val EXTRA_SCREENSHOT_SESSION_ID = "screenshotSessionId"
@@ -152,7 +153,24 @@ class MainActivity : FlutterActivity() {
                    )
                    result.success(response?.getStringArrayList("candidates") ?: arrayListOf<String>())
                }
-                "ackRecognitionCandidate" -> acknowledgeRecognition(call, result)
+                "ackRecognitionCandidate" -> acknowledgeRecognition(call, result)
+                "listRecognitionBatches" -> {
+                    val response = RecognitionBridge.call(
+                        this,
+                        RecognitionBridgeProvider.METHOD_BATCHES,
+                    )
+                    result.success(response?.getStringArrayList("batches") ?: arrayListOf<String>())
+                }
+                "restoreDroppedRecognition" -> {
+                    val response = RecognitionBridge.call(
+                        this,
+                        RecognitionBridgeProvider.METHOD_RESTORE_DROPPED,
+                        extras = Bundle().apply {
+                            putString("candidateId", call.argument<String>("candidateId"))
+                        },
+                    )
+                    result.success(response?.getBoolean("success") == true)
+                }
                "openAccessibilitySettings" -> {
                    startActivity(Intent(Settings.ACTION_ACCESSIBILITY_SETTINGS))
                    result.success(true)
@@ -240,18 +258,22 @@ class MainActivity : FlutterActivity() {
                    dispatchPendingScreenshot()
                }
            }
-            ACTION_RECOGNITION_CONFIRM,
-            ACTION_RECOGNITION_UNDO,
-            ACTION_RECOGNITION_EDIT -> {
+            ACTION_RECOGNITION_CONFIRM,
+            ACTION_RECOGNITION_UNDO,
+            ACTION_RECOGNITION_EDIT,
+            ACTION_RECOGNITION_BATCH_REVIEW -> {
                pendingRecognitionAction = mapOf(
                    "action" to incoming.getStringExtra(EXTRA_ACTION),
                    "candidateId" to incoming.getStringExtra(
                        RecognitionCoordinator.EXTRA_CANDIDATE_ID,
                    ),
-                    "transactionId" to incoming.getLongExtra(
+                    "transactionId" to incoming.getLongExtra(
                        EXTRA_TRANSACTION_ID,
                        Long.MIN_VALUE,
-                    ).takeIf { it != Long.MIN_VALUE },
+                    ).takeIf { it != Long.MIN_VALUE },
+                    "batchId" to incoming.getStringExtra(
+                        RecognitionCoordinator.EXTRA_BATCH_ID,
+                    ),
                )
                dispatchPendingRecognitionAction()
            }
@@ -810,9 +832,12 @@ class MainActivity : FlutterActivity() {
        override fun onReceive(context: Context?, intent: Intent?) {
            pendingRecognitionAction = mapOf(
                "action" to "ready",
-                "candidateId" to intent?.getStringExtra(
+                "candidateId" to intent?.getStringExtra(
                    RecognitionCoordinator.EXTRA_CANDIDATE_ID,
-                ),
+                ),
+                "batchId" to intent?.getStringExtra(
+                    RecognitionCoordinator.EXTRA_BATCH_ID,
+                ),
            )
            dispatchPendingRecognitionAction()
        }
@@ -0,0 +1,92 @@
+package com.nx.miaoji
+
+import android.content.Context
+import android.util.Log
+import java.net.HttpURLConnection
+import java.net.URL
+import java.util.UUID
+
+class RecognitionBatchProcessor(private val context: Context) {
+    fun process(batch: PendingRecognitionBatch): BatchProcessResult {
+        return try {
+            val settings = RecognitionSettings.snapshot(context)
+            val token = RecognitionSettings.runtimeToken(context)
+            val baseUrl = settings.baseUrl?.trimEnd('/')
+            if (!settings.aiScreenshot || !settings.aiAllowed || !settings.hasAccount ||
+                token.isNullOrBlank() || baseUrl.isNullOrBlank()
+            ) {
+                return BatchProcessResult.Fallback("AI 截图补全不可用")
+            }
+            val response = upload(
+                "$baseUrl/api/parse/recognition-batch",
+                token,
+                batch,
+            )
+            if (response.code == HttpURLConnection.HTTP_FORBIDDEN &&
+                response.body.contains("AI_PERMISSION_DENIED")
+            ) {
+                RecognitionSettings.disableRuntimeAi(context)
+            }
+            if (response.code !in 200..299) {
+                Log.w(TAG, "Recognition batch failed status=${response.code} batchId=${batch.id}")
+                BatchProcessResult.Fallback("AI 批次对账失败（${response.code}）")
+            } else {
+                BatchProcessResult.Success(response.body)
+            }
+        } catch (error: Exception) {
+            Log.w(TAG, "Recognition batch unavailable batchId=${batch.id}", error)
+            BatchProcessResult.Fallback("AI 批次对账超时或网络不可用")
+        } finally {
+            batch.images.forEach { it.bytes.fill(0) }
+        }
+    }
+
+    private fun upload(
+        endpoint: String,
+        token: String,
+        batch: PendingRecognitionBatch,
+    ): HttpResponse {
+        val boundary = "----JizhiBatch${UUID.randomUUID()}"
+        val connection = URL(endpoint).openConnection() as HttpURLConnection
+        connection.connectTimeout = 10_000
+        connection.readTimeout = 120_000
+        connection.requestMethod = "POST"
+        connection.doOutput = true
+        connection.setRequestProperty("Authorization", "Bearer $token")
+        connection.setRequestProperty("Content-Type", "multipart/form-data; boundary=$boundary")
+        connection.outputStream.use { output ->
+            output.write("--$boundary\r\n".toByteArray())
+            output.write("Content-Disposition: form-data; name=\"manifest\"\r\n".toByteArray())
+            output.write("Content-Type: application/json; charset=utf-8\r\n\r\n".toByteArray())
+            output.write(batch.manifest.toByteArray(Charsets.UTF_8))
+            output.write("\r\n".toByteArray())
+            batch.images.forEach { image ->
+                output.write("--$boundary\r\n".toByteArray())
+                output.write(
+                    "Content-Disposition: form-data; name=\"files\"; filename=\"${image.evidenceId}.jpg\"\r\n"
+                        .toByteArray(),
+                )
+                output.write("Content-Type: image/jpeg\r\n\r\n".toByteArray())
+                output.write(image.bytes)
+                output.write("\r\n".toByteArray())
+            }
+            output.write("--$boundary--\r\n".toByteArray())
+        }
+        val code = connection.responseCode
+        val stream = if (code in 200..299) connection.inputStream else connection.errorStream
+        val body = stream?.bufferedReader(Charsets.UTF_8)?.use { it.readText() }.orEmpty()
+        connection.disconnect()
+        return HttpResponse(code, body)
+    }
+
+    private data class HttpResponse(val code: Int, val body: String)
+
+    companion object {
+        private const val TAG = "JizhiRecognition"
+    }
+}
+
+sealed interface BatchProcessResult {
+    data class Success(val responseBody: String) : BatchProcessResult
+    data class Fallback(val reason: String) : BatchProcessResult
+}
@@ -73,6 +73,19 @@ class RecognitionBridgeProvider : ContentProvider() {
                )
                putBoolean("success", candidate != null)
            }
+            METHOD_BATCHES -> Bundle().apply {
+                putStringArrayList(
+                    "batches",
+                    ArrayList(RecognitionCoordinator.get(appContext).recentBatches()),
+                )
+            }
+            METHOD_RESTORE_DROPPED -> Bundle().apply {
+                putBoolean(
+                    "success",
+                    RecognitionCoordinator.get(appContext)
+                        .restoreDropped(extras?.getString("candidateId").orEmpty()) != null,
+                )
+            }
            else -> super.call(method, arg, extras) ?: Bundle()
        }
    }
@@ -132,6 +145,8 @@ class RecognitionBridgeProvider : ContentProvider() {
        const val METHOD_SCREENSHOT_RESULT = "screenshotResult"
        const val METHOD_DRAIN = "drain"
        const val METHOD_ACK = "ack"
+        const val METHOD_BATCHES = "batches"
+        const val METHOD_RESTORE_DROPPED = "restoreDropped"
    }
 }

@@ -4,19 +4,101 @@ import android.content.Context
 import android.content.Intent
 import android.os.Handler
 import android.os.HandlerThread
+import android.os.Looper
 import android.util.Log

 class RecognitionCoordinator private constructor(private val context: Context) {
    private val store = RecognitionStore(context)
+    private val batchProcessor = RecognitionBatchProcessor(context)
    private val thread = HandlerThread("jizhi-recognition").apply { start() }
    private val handler = Handler(thread.looper)
+    private val mainHandler = Handler(Looper.getMainLooper())
+    private val batchRunnable = Runnable { processDueBatches() }

-    fun submit(signal: PaymentSignal) {
+    init {
        handler.post {
-            runCatching {
-                val id = store.upsert(signal)
-                handler.postDelayed({ finalize(id) }, 1_650L)
-            }.onFailure { Log.e(TAG, "Unable to store recognition signal", it) }
+            store.recoverInterruptedBatches()
+            scheduleBatchProcessing()
+        }
+    }
+
+    fun submit(
+        signal: PaymentSignal,
+        evidenceImage: ByteArray? = null,
+        onStored: ((StoredSubmission) -> Unit)? = null,
+    ) {
+        handler.post {
+            try {
+                val batchMode = RecognitionSettings.snapshot(context).let {
+                    it.aiScreenshot && it.aiAllowed && it.hasAccount
+                }
+                val submission = store.upsert(signal, batchMode)
+                if (submission.batchId != null && evidenceImage != null) {
+                    store.addBatchImage(
+                        submission.batchId,
+                        submission.candidateId,
+                        signal.flowSessionId,
+                        signal.packageName,
+                        System.currentTimeMillis(),
+                        evidenceImage,
+                    )
+                }
+                if (submission.batchId == null) {
+                    handler.postDelayed({ finalize(submission.candidateId) }, 1_650L)
+                } else {
+                    scheduleBatchProcessing()
+                }
+                onStored?.let { callback ->
+                    mainHandler.post { callback(submission) }
+                }
+            } catch (error: Exception) {
+                Log.e(TAG, "Unable to store recognition signal", error)
+            } finally {
+                evidenceImage?.fill(0)
+            }
+        }
+    }
+
+    fun attachBatchImage(
+        submission: StoredSubmission,
+        signal: PaymentSignal,
+        image: ByteArray,
+    ) {
+        val batchId = submission.batchId
+        if (batchId == null) {
+            image.fill(0)
+            return
+        }
+        handler.post {
+            try {
+                store.addBatchImage(
+                    batchId,
+                    submission.candidateId,
+                    signal.flowSessionId,
+                    signal.packageName,
+                    System.currentTimeMillis(),
+                    image,
+                )
+                scheduleBatchProcessing()
+            } finally {
+                image.fill(0)
+            }
+        }
+    }
+
+    fun submitEvidenceOnly(
+        packageName: String,
+        flowSessionId: String?,
+        capturedAt: Long,
+        image: ByteArray,
+    ) {
+        handler.post {
+            try {
+                store.addEvidenceOnly(packageName, flowSessionId, capturedAt, image)
+                scheduleBatchProcessing()
+            } finally {
+                image.fill(0)
+            }
        }
    }

@@ -24,12 +106,18 @@ class RecognitionCoordinator private constructor(private val context: Context) {

    fun acknowledge(id: String, state: String, transactionId: Long?): StoredCandidate? {
        val candidate = store.acknowledge(id, state, transactionId)
-        if (candidate != null && state == "imported") {
+        if (candidate != null && state == "imported" && candidate.batchId == null) {
            RecognitionNotifier.showImported(context, candidate)
        }
        return candidate
    }

+    fun recentBatches(): List<String> = store.recentBatches()
+
+    fun restoreDropped(candidateId: String): StoredCandidate? {
+        return store.restoreDropped(candidateId)
+    }
+
    fun latestStatus(): String? = store.latestStatus()

    private fun finalize(id: String) {
@@ -42,9 +130,48 @@ class RecognitionCoordinator private constructor(private val context: Context) {
        RecognitionNotifier.showReady(context, candidate)
    }

+    private fun scheduleBatchProcessing() {
+        handler.removeCallbacks(batchRunnable)
+        val dueAt = store.nextBatchDueAt() ?: return
+        handler.postDelayed(batchRunnable, (dueAt - System.currentTimeMillis()).coerceAtLeast(0L))
+    }
+
+    private fun processDueBatches() {
+        while (true) {
+            val batch = store.claimDueBatch(System.currentTimeMillis()) ?: break
+            val fallback = when (val result = batchProcessor.process(batch)) {
+                is BatchProcessResult.Success -> {
+                    val applied = runCatching {
+                        store.applyBatchResponse(batch.id, result.responseBody)
+                    }.onFailure {
+                        Log.e(TAG, "Unable to apply recognition batch ${batch.id}", it)
+                    }.getOrDefault(false)
+                    if (applied) {
+                        false
+                    } else {
+                        store.fallbackBatch(batch.id, "AI 返回结果无法应用")
+                        true
+                    }
+                }
+                is BatchProcessResult.Fallback -> {
+                    store.fallbackBatch(batch.id, result.reason)
+                    true
+                }
+            }
+            context.sendBroadcast(
+                Intent(ACTION_READY)
+                    .setPackage(context.packageName)
+                    .putExtra(EXTRA_BATCH_ID, batch.id),
+            )
+            RecognitionNotifier.showBatchReady(context, batch.id, fallback)
+        }
+        scheduleBatchProcessing()
+    }
+
    companion object {
        const val ACTION_READY = "com.nx.miaoji.RECOGNITION_READY"
        const val EXTRA_CANDIDATE_ID = "candidateId"
+        const val EXTRA_BATCH_ID = "batchId"
        private const val TAG = "JizhiRecognition"

        @Volatile
@@ -68,6 +68,26 @@ object RecognitionNotifier {
        notify(context, candidate.id.hashCode(), notification)
    }

+    fun showBatchReady(context: Context, batchId: String, fallback: Boolean) {
+        val intent = Intent(context, MainActivity::class.java).apply {
+            addFlags(Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_SINGLE_TOP)
+            putExtra(MainActivity.EXTRA_ACTION, MainActivity.ACTION_RECOGNITION_BATCH_REVIEW)
+            putExtra(RecognitionCoordinator.EXTRA_BATCH_ID, batchId)
+        }
+        val notification = Notification.Builder(context, ensureChannel(context))
+            .setSmallIcon(R.mipmap.ic_launcher)
+            .setContentTitle(if (fallback) "本地识别结果已就绪" else "AI 批次对账已完成")
+            .setContentText(
+                if (fallback) "AI 暂时不可用，已按本地结果处理"
+                else "点按查看本批次的保留、修正、补全和剔除结果",
+            )
+            .setAutoCancel(true)
+            .setContentIntent(pendingActivity(context, batchId.hashCode(), intent))
+            .setCategory(Notification.CATEGORY_STATUS)
+            .build()
+        notify(context, batchId.hashCode(), notification)
+    }
+
    private fun actionIntent(context: Context, candidate: StoredCandidate, action: String) =
        Intent(context, MainActivity::class.java).apply {
            addFlags(Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_SINGLE_TOP)
@@ -81,7 +81,12 @@ object RecognitionSettings {
        if (token.isNullOrBlank()) {
            editor.remove(KEY_TOKEN)
        } else {
-            val encrypted = NativeCrypto.encrypt(token.toByteArray(Charsets.UTF_8))
+            val tokenBytes = token.toByteArray(Charsets.UTF_8)
+            val encrypted = try {
+                NativeCrypto.encrypt(tokenBytes)
+            } finally {
+                tokenBytes.fill(0)
+            }
            if (encrypted != null) editor.putString(KEY_TOKEN, encrypted)
        }
        editor.apply()
@@ -97,7 +102,12 @@ object RecognitionSettings {
    fun runtimeToken(context: Context): String? {
        val encoded = context.getSharedPreferences(PREFS, Context.MODE_PRIVATE)
            .getString(KEY_TOKEN, null) ?: return null
-        return NativeCrypto.decrypt(encoded)?.toString(Charsets.UTF_8)
+        val decrypted = NativeCrypto.decrypt(encoded) ?: return null
+        return try {
+            decrypted.toString(Charsets.UTF_8)
+        } finally {
+            decrypted.fill(0)
+        }
    }

    fun statusJson(context: Context): String {
@@ -6,6 +6,8 @@ import android.database.Cursor
 import android.database.sqlite.SQLiteDatabase
 import android.database.sqlite.SQLiteOpenHelper
 import org.json.JSONObject
+import org.json.JSONArray
+import java.time.Instant
 import java.util.UUID
 import kotlin.math.abs

@@ -15,10 +17,29 @@ data class StoredCandidate(
    val state: String,
    val transactionId: Long?,
    val json: String,
+    val batchId: String? = null,
+)
+
+data class StoredSubmission(val candidateId: String, val batchId: String?)
+
+data class PendingBatchImage(
+    val evidenceId: String,
+    val candidateId: String?,
+    val flowSessionId: String?,
+    val packageName: String,
+    val capturedAt: Long,
+    val bytes: ByteArray,
+)
+
+data class PendingRecognitionBatch(
+    val id: String,
+    val openedAt: Long,
+    val manifest: String,
+    val images: List<PendingBatchImage>,
 )

 class RecognitionStore(context: Context) :
-    SQLiteOpenHelper(context, "recognition_queue.db", null, 2) {
+    SQLiteOpenHelper(context, "recognition_queue.db", null, 3) {
    override fun onCreate(db: SQLiteDatabase) {
        db.execSQL(
            """
@@ -39,7 +60,11 @@ class RecognitionStore(context: Context) :
              available_at INTEGER NOT NULL,
              first_seen INTEGER NOT NULL,
              updated_at INTEGER NOT NULL,
-              transaction_id INTEGER
+              transaction_id INTEGER,
+              batch_id TEXT,
+              ai_action TEXT,
+              ai_reason TEXT,
+              original_payload_encrypted TEXT
            )
            """.trimIndent(),
        )
@@ -64,6 +89,7 @@ class RecognitionStore(context: Context) :
        db.execSQL(
            "CREATE INDEX ix_recognition_state ON candidates(state, available_at)",
        )
+        createBatchTables(db)
    }

    override fun onUpgrade(db: SQLiteDatabase, oldVersion: Int, newVersion: Int) {
@@ -73,10 +99,50 @@ class RecognitionStore(context: Context) :
                    "ON candidates(state, available_at)",
            )
        }
+        if (oldVersion < 3) {
+            db.execSQL("ALTER TABLE candidates ADD COLUMN batch_id TEXT")
+            db.execSQL("ALTER TABLE candidates ADD COLUMN ai_action TEXT")
+            db.execSQL("ALTER TABLE candidates ADD COLUMN ai_reason TEXT")
+            db.execSQL("ALTER TABLE candidates ADD COLUMN original_payload_encrypted TEXT")
+            createBatchTables(db)
+        }
+    }
+
+    private fun createBatchTables(db: SQLiteDatabase) {
+        db.execSQL(
+            """
+            CREATE TABLE IF NOT EXISTS recognition_batches (
+              id TEXT PRIMARY KEY,
+              state TEXT NOT NULL,
+              opened_at INTEGER NOT NULL,
+              last_seen INTEGER NOT NULL,
+              flush_at INTEGER NOT NULL,
+              hard_deadline INTEGER NOT NULL,
+              completed_at INTEGER,
+              summary_json TEXT,
+              failure_reason TEXT
+            )
+            """.trimIndent(),
+        )
+        db.execSQL(
+            """
+            CREATE TABLE IF NOT EXISTS batch_images (
+              evidence_id TEXT PRIMARY KEY,
+              batch_id TEXT NOT NULL,
+              candidate_id TEXT,
+              flow_session_id TEXT,
+              package_name TEXT NOT NULL,
+              captured_at INTEGER NOT NULL,
+              image_encrypted TEXT NOT NULL
+            )
+            """.trimIndent(),
+        )
+        db.execSQL("CREATE INDEX IF NOT EXISTS ix_candidates_batch ON candidates(batch_id, state)")
+        db.execSQL("CREATE INDEX IF NOT EXISTS ix_batches_due ON recognition_batches(state, flush_at)")
    }

    @Synchronized
-    fun upsert(signal: PaymentSignal): String {
+    fun upsert(signal: PaymentSignal, batchMode: Boolean = false): StoredSubmission {
        val now = System.currentTimeMillis()
        val channelBit = channelBit(signal.channel)
        val sourceHash = PaymentParser.sha256(signal.sourceEventId)
@@ -88,7 +154,8 @@ class RecognitionStore(context: Context) :
            ).use { cursor ->
                if (cursor.moveToFirst()) {
                    writableDatabase.setTransactionSuccessful()
-                    return cursor.getString(0)
+                    val candidateId = cursor.getString(0)
+                    return StoredSubmission(candidateId, batchIdForCandidate(candidateId))
                }
            }

@@ -98,15 +165,14 @@ class RecognitionStore(context: Context) :
            val orderStrongKey = signal.orderId?.takeIf { it.isNotBlank() }?.let {
                PaymentParser.sha256("${signal.packageName}|$it")
            }
-            val flowStrongKey = signal.flowSessionId?.takeIf { it.isNotBlank() }?.let {
-                PaymentParser.sha256("${signal.packageName}|flow|$it")
-            }
+            val flowStrongKey = flowStrongKeyFor(signal)
            val strongKey = orderStrongKey ?: flowStrongKey
            val clientRequestId = clientRequestIdFor(signal)
            val signalHigh = signal.channel in setOf("accessibility", "local_ocr") &&
                signal.evidenceConfidence == "high"
            val existing = findMergeCandidate(signal, channelBit, merchantHash, strongKey, now)
                ?: findByClientRequestId(clientRequestId)
+            val batchId = existing?.batchId ?: if (batchMode) activeBatch(now) else null
            val id: String
            if (existing != null) {
                id = existing.id
@@ -128,6 +194,10 @@ class RecognitionStore(context: Context) :
                        if (high && existing.state == "pending_confirm") put("state", "auto_ready")
                        put("updated_at", now)
                        put("available_at", now + MERGE_DELAY_MS)
+                        if (batchId != null) {
+                            put("batch_id", batchId)
+                            put("state", "batch_collecting")
+                        }
                    },
                    "id = ?",
                    arrayOf(id),
@@ -151,11 +221,12 @@ class RecognitionStore(context: Context) :
                        put("known_template", if (signal.knownTemplate) 1 else 0)
                        put("occurred_at", signal.occurredAtEpochMs)
                        put("payload_encrypted", encryptPayload(payload))
-                        put("state", "pending_merge")
+                        put("state", if (batchId == null) "pending_merge" else "batch_collecting")
                        put("high_confidence", if (high) 1 else 0)
                        put("available_at", now + MERGE_DELAY_MS)
                        put("first_seen", now)
                        put("updated_at", now)
+                        if (batchId != null) put("batch_id", batchId)
                    },
                )
            }
@@ -169,8 +240,9 @@ class RecognitionStore(context: Context) :
                    put("created_at", now)
                },
            )
+            if (batchId != null) touchBatch(batchId, now)
            writableDatabase.setTransactionSuccessful()
-            return id
+            return StoredSubmission(id, batchId)
        } finally {
            writableDatabase.endTransaction()
        }
@@ -184,6 +256,563 @@ class RecognitionStore(context: Context) :
            if (cursor.moveToFirst()) row(cursor) else null
        }

+    private fun batchIdForCandidate(candidateId: String): String? = readableDatabase.rawQuery(
+        "SELECT batch_id FROM candidates WHERE id = ? LIMIT 1",
+        arrayOf(candidateId),
+    ).use { cursor ->
+        if (!cursor.moveToFirst() || cursor.isNull(0)) null else cursor.getString(0)
+    }
+
+    private fun activeBatch(now: Long): String {
+        readableDatabase.rawQuery(
+            """
+            SELECT b.id,
+              (SELECT COUNT(*) FROM candidates c WHERE c.batch_id = b.id) +
+              (SELECT COUNT(*) FROM batch_images i WHERE i.batch_id = b.id AND i.candidate_id IS NULL)
+            FROM recognition_batches b
+            WHERE b.state = 'collecting' AND b.hard_deadline > ?
+            ORDER BY b.opened_at DESC LIMIT 1
+            """.trimIndent(),
+            arrayOf(now.toString()),
+        ).use { cursor ->
+            if (cursor.moveToFirst() && cursor.getInt(1) < MAX_BATCH_ITEMS) {
+                return cursor.getString(0)
+            }
+        }
+        val id = UUID.randomUUID().toString()
+        writableDatabase.insertOrThrow(
+            "recognition_batches",
+            null,
+            ContentValues().apply {
+                put("id", id)
+                put("state", "collecting")
+                put("opened_at", now)
+                put("last_seen", now)
+                put("flush_at", now + BATCH_IDLE_MS)
+                put("hard_deadline", now + BATCH_HARD_LIMIT_MS)
+            },
+        )
+        return id
+    }
+
+    private fun touchBatch(batchId: String, now: Long) {
+        writableDatabase.execSQL(
+            """
+            UPDATE recognition_batches
+            SET last_seen = ?,
+                flush_at = MIN(hard_deadline, ?)
+            WHERE id = ? AND state = 'collecting'
+            """.trimIndent(),
+            arrayOf<Any>(now, now + BATCH_IDLE_MS, batchId),
+        )
+        val itemCount = readableDatabase.rawQuery(
+            """
+            SELECT
+              (SELECT COUNT(*) FROM candidates WHERE batch_id = ?) +
+              (SELECT COUNT(*) FROM batch_images WHERE batch_id = ? AND candidate_id IS NULL)
+            """.trimIndent(),
+            arrayOf(batchId, batchId),
+        ).use { cursor -> if (cursor.moveToFirst()) cursor.getInt(0) else 0 }
+        if (itemCount >= MAX_BATCH_ITEMS) {
+            writableDatabase.execSQL(
+                "UPDATE recognition_batches SET flush_at = ? WHERE id = ?",
+                arrayOf<Any>(now, batchId),
+            )
+        }
+    }
+
+    @Synchronized
+    fun addBatchImage(
+        batchId: String,
+        candidateId: String?,
+        flowSessionId: String?,
+        packageName: String,
+        capturedAt: Long,
+        image: ByteArray,
+    ): String? {
+        if (image.isEmpty() || image.size > MAX_BATCH_IMAGE_BYTES) return null
+        val state = readableDatabase.rawQuery(
+            "SELECT state FROM recognition_batches WHERE id = ? LIMIT 1",
+            arrayOf(batchId),
+        ).use { cursor -> if (cursor.moveToFirst()) cursor.getString(0) else null }
+        if (state != "collecting") return null
+        val existing = readableDatabase.rawQuery(
+            """
+            SELECT evidence_id FROM batch_images
+            WHERE batch_id = ? AND (
+              (? IS NOT NULL AND candidate_id = ?) OR
+              (? IS NULL AND candidate_id IS NULL AND flow_session_id = ?)
+            ) LIMIT 1
+            """.trimIndent(),
+            arrayOf(batchId, candidateId, candidateId, candidateId, flowSessionId),
+        ).use { cursor -> if (cursor.moveToFirst()) cursor.getString(0) else null }
+        if (existing != null) return existing
+        val encrypted = NativeCrypto.encrypt(image) ?: return null
+        val evidenceId = UUID.randomUUID().toString()
+        writableDatabase.insertOrThrow(
+            "batch_images",
+            null,
+            ContentValues().apply {
+                put("evidence_id", evidenceId)
+                put("batch_id", batchId)
+                if (candidateId == null) putNull("candidate_id") else put("candidate_id", candidateId)
+                if (flowSessionId == null) putNull("flow_session_id") else put("flow_session_id", flowSessionId)
+                put("package_name", packageName)
+                put("captured_at", capturedAt)
+                put("image_encrypted", encrypted)
+            },
+        )
+        touchBatch(batchId, capturedAt)
+        return evidenceId
+    }
+
+    @Synchronized
+    fun addEvidenceOnly(
+        packageName: String,
+        flowSessionId: String?,
+        capturedAt: Long,
+        image: ByteArray,
+    ): String? {
+        writableDatabase.beginTransaction()
+        return try {
+            val batchId = activeBatch(capturedAt)
+            addBatchImage(batchId, null, flowSessionId, packageName, capturedAt, image)
+            writableDatabase.setTransactionSuccessful()
+            batchId
+        } finally {
+            writableDatabase.endTransaction()
+        }
+    }
+
+    @Synchronized
+    fun nextBatchDueAt(): Long? = readableDatabase.rawQuery(
+        "SELECT MIN(flush_at) FROM recognition_batches WHERE state = 'collecting'",
+        null,
+    ).use { cursor ->
+        if (!cursor.moveToFirst() || cursor.isNull(0)) null else cursor.getLong(0)
+    }
+
+    @Synchronized
+    fun claimDueBatch(now: Long): PendingRecognitionBatch? {
+        val batch = readableDatabase.rawQuery(
+            """
+            SELECT id, opened_at FROM recognition_batches
+            WHERE state = 'collecting' AND (flush_at <= ? OR hard_deadline <= ?)
+            ORDER BY opened_at LIMIT 1
+            """.trimIndent(),
+            arrayOf(now.toString(), now.toString()),
+        ).use { cursor ->
+            if (!cursor.moveToFirst()) null else cursor.getString(0) to cursor.getLong(1)
+        } ?: return null
+        writableDatabase.update(
+            "recognition_batches",
+            ContentValues().apply { put("state", "processing") },
+            "id = ? AND state = 'collecting'",
+            arrayOf(batch.first),
+        )
+
+        val images = readableDatabase.rawQuery(
+            "SELECT * FROM batch_images WHERE batch_id = ? ORDER BY captured_at",
+            arrayOf(batch.first),
+        ).use { cursor ->
+            buildList {
+                while (cursor.moveToNext()) {
+                    val bytes = NativeCrypto.decrypt(cursor.getString(cursor.getColumnIndexOrThrow("image_encrypted")))
+                        ?: continue
+                    val candidateIndex = cursor.getColumnIndexOrThrow("candidate_id")
+                    val flowIndex = cursor.getColumnIndexOrThrow("flow_session_id")
+                    add(
+                        PendingBatchImage(
+                            evidenceId = cursor.getString(cursor.getColumnIndexOrThrow("evidence_id")),
+                            candidateId = if (cursor.isNull(candidateIndex)) null else cursor.getString(candidateIndex),
+                            flowSessionId = if (cursor.isNull(flowIndex)) null else cursor.getString(flowIndex),
+                            packageName = cursor.getString(cursor.getColumnIndexOrThrow("package_name")),
+                            capturedAt = cursor.getLong(cursor.getColumnIndexOrThrow("captured_at")),
+                            bytes = bytes,
+                        ),
+                    )
+                }
+            }
+        }
+        val evidenceByCandidate = images.filter { it.candidateId != null }.groupBy { it.candidateId }
+        val candidates = JSONArray()
+        readableDatabase.rawQuery(
+            "SELECT * FROM candidates WHERE batch_id = ? AND state = 'batch_collecting' ORDER BY occurred_at, first_seen",
+            arrayOf(batch.first),
+        ).use { cursor ->
+            while (cursor.moveToNext()) {
+                val payload = decryptPayload(cursor.getString(cursor.getColumnIndexOrThrow("payload_encrypted")))
+                    ?: continue
+                candidates.put(
+                    JSONObject()
+                        .put("candidateId", cursor.getString(cursor.getColumnIndexOrThrow("id")))
+                        .put("clientRequestId", cursor.getString(cursor.getColumnIndexOrThrow("client_request_id")))
+                        .put("flowSessionId", payload.optString("flowSessionId").takeIf(String::isNotBlank))
+                        .put("packageName", payload.optString("packageName"))
+                        .put("type", payload.optString("type"))
+                        .put("amount", payload.optDouble("amount"))
+                        .put("merchant", payload.optString("merchant").takeIf(String::isNotBlank))
+                        .put("orderId", payload.optString("orderId").takeIf(String::isNotBlank))
+                        .put("occurredAt", Instant.ofEpochMilli(payload.optLong("occurredAtEpochMs")).toString())
+                        .put("recognitionKind", payload.optString("recognitionKind", "payment"))
+                        .put("categoryHint", payload.optString("categoryHint").takeIf(String::isNotBlank))
+                        .put("confidence", if (cursor.getInt(cursor.getColumnIndexOrThrow("high_confidence")) == 1) "auto" else "confirm")
+                        .put("sourceText", payload.optString("sourceText").takeIf(String::isNotBlank))
+                        .put(
+                            "evidenceIds",
+                            JSONArray(evidenceByCandidate[cursor.getString(cursor.getColumnIndexOrThrow("id"))]
+                                .orEmpty().map(PendingBatchImage::evidenceId)),
+                        ),
+                )
+            }
+        }
+        val evidence = JSONArray(images.map { image ->
+            JSONObject()
+                .put("evidenceId", image.evidenceId)
+                .put("candidateId", image.candidateId)
+                .put("flowSessionId", image.flowSessionId)
+                .put("packageName", image.packageName)
+                .put("capturedAt", Instant.ofEpochMilli(image.capturedAt).toString())
+        })
+        val manifest = JSONObject()
+            .put("batchId", batch.first)
+            .put("openedAt", Instant.ofEpochMilli(batch.second).toString())
+            .put("candidates", candidates)
+            .put("evidence", evidence)
+            .toString()
+        return PendingRecognitionBatch(batch.first, batch.second, manifest, images)
+    }
+
+    @Synchronized
+    fun recoverInterruptedBatches() {
+        val now = System.currentTimeMillis()
+        writableDatabase.execSQL(
+            "UPDATE recognition_batches SET state = 'collecting', flush_at = ? WHERE state = 'processing'",
+            arrayOf(now),
+        )
+        cleanupBatchHistory(now)
+    }
+
+    @Synchronized
+    fun applyBatchResponse(batchId: String, responseBody: String): Boolean {
+        val root = runCatching { JSONObject(responseBody) }.getOrNull() ?: return false
+        if (root.optString("batchId") != batchId) return false
+        val actions = root.optJSONArray("actions") ?: return false
+        val now = System.currentTimeMillis()
+        writableDatabase.beginTransaction()
+        return try {
+            val seenCandidates = HashSet<String>()
+            var kept = 0
+            var updated = 0
+            var created = 0
+            var dropped = 0
+            for (index in 0 until actions.length()) {
+                val action = actions.optJSONObject(index) ?: continue
+                when (action.optString("action")) {
+                    "keep", "update", "drop" -> {
+                        val candidateId = action.optString("candidateId")
+                        if (candidateId.isBlank() || !seenCandidates.add(candidateId)) continue
+                        val row = readableDatabase.rawQuery(
+                            "SELECT * FROM candidates WHERE id = ? AND batch_id = ? LIMIT 1",
+                            arrayOf(candidateId, batchId),
+                        ).use { cursor -> if (cursor.moveToFirst()) row(cursor) else null } ?: continue
+                        val original = JSONObject(row.payload.toString())
+                        val kind = action.optString("action")
+                        val payload = JSONObject(row.payload.toString())
+                        if (kind == "update") applyActionFields(payload, action)
+                        payload.put("sourceOverride", "recognition_ai")
+                        val reason = action.optString("reason", "AI 对账")
+                        writableDatabase.update(
+                            "candidates",
+                            ContentValues().apply {
+                                put("original_payload_encrypted", encryptPayload(original))
+                                put("payload_encrypted", encryptPayload(payload))
+                                put("amount_cents", kotlin.math.round(payload.optDouble("amount") * 100).toLong())
+                                put("direction", payload.optString("type"))
+                                put("merchant_hash", PaymentParser.sha256(payload.optString("merchant").lowercase()))
+                                put("occurred_at", payload.optLong("occurredAtEpochMs"))
+                                put("channel_mask", row.channelMask or channelBit("recognition_ai"))
+                                put("ai_action", kind)
+                                put("ai_reason", reason.take(80))
+                                put("state", if (kind == "drop") "ai_dropped" else "auto_ready")
+                                put("high_confidence", 1)
+                                put("updated_at", now)
+                            },
+                            "id = ? AND batch_id = ?",
+                            arrayOf(candidateId, batchId),
+                        )
+                        when (kind) {
+                            "keep" -> kept += 1
+                            "update" -> updated += 1
+                            "drop" -> dropped += 1
+                        }
+                    }
+                    "create" -> {
+                        val evidenceId = action.optString("evidenceId")
+                        if (evidenceId.isBlank()) continue
+                        val image = readableDatabase.rawQuery(
+                            "SELECT * FROM batch_images WHERE evidence_id = ? AND batch_id = ? AND candidate_id IS NULL LIMIT 1",
+                            arrayOf(evidenceId, batchId),
+                        ).use { cursor ->
+                            if (!cursor.moveToFirst()) null else Triple(
+                                cursor.getString(cursor.getColumnIndexOrThrow("package_name")),
+                                cursor.getLong(cursor.getColumnIndexOrThrow("captured_at")),
+                                cursor.getColumnIndexOrThrow("flow_session_id").let { flowIndex ->
+                                    if (cursor.isNull(flowIndex)) null else cursor.getString(flowIndex)
+                                },
+                            )
+                        } ?: continue
+                        val amount = action.optDouble("amount", 0.0)
+                        val type = action.optString("type")
+                        if (amount <= 0 || type !in setOf("income", "expense")) continue
+                        val candidateId = UUID.randomUUID().toString()
+                        val requestId = "recognition-" + PaymentParser.sha256("$batchId|create|$evidenceId").take(52)
+                        val payload = JSONObject()
+                            .put("packageName", image.first)
+                            .put("appName", PaymentParser.appName(image.first))
+                            .put("type", type)
+                            .put("amount", amount)
+                            .put("merchant", action.optionalString("note"))
+                            .put("orderId", JSONObject.NULL)
+                            .put("occurredAtEpochMs", action.optionalInstantEpoch("occurredAt") ?: image.second)
+                            .put("sourceText", "AI 批次补全 · ${PaymentParser.appName(image.first)}")
+                            .put("flowSessionId", image.third)
+                            .put("evidenceConfidence", "high")
+                            .put("recognitionKind", "payment")
+                            .put("categoryHint", action.optionalString("categoryName"))
+                            .put("categoryId", action.optLong("categoryId").takeIf { it > 0 })
+                            .put("amountSource", "ai_batch")
+                            .put("resultFingerprint", PaymentParser.sha256("$batchId|$evidenceId"))
+                            .put("note", action.optionalString("note") ?: PaymentParser.appName(image.first))
+                            .put("paymentMethod", action.optionalString("paymentMethod"))
+                            .put("sourceOverride", "recognition_ai")
+                        writableDatabase.insertOrThrow(
+                            "candidates",
+                            null,
+                            ContentValues().apply {
+                                put("id", candidateId)
+                                put("client_request_id", requestId)
+                                put("package_name", image.first)
+                                put("amount_cents", kotlin.math.round(amount * 100).toLong())
+                                put("direction", type)
+                                put("merchant_hash", PaymentParser.sha256(payload.optString("merchant").lowercase()))
+                                put("strong_key", PaymentParser.sha256("$batchId|create|$evidenceId"))
+                                put("channel_mask", channelBit("recognition_ai"))
+                                put("known_template", 0)
+                                put("occurred_at", payload.optLong("occurredAtEpochMs"))
+                                put("payload_encrypted", encryptPayload(payload))
+                                put("state", "auto_ready")
+                                put("high_confidence", 1)
+                                put("available_at", now)
+                                put("first_seen", now)
+                                put("updated_at", now)
+                                put("batch_id", batchId)
+                                put("ai_action", "create")
+                                put("ai_reason", action.optString("reason", "AI 补全").take(80))
+                            },
+                        )
+                        created += 1
+                    }
+                }
+            }
+            // The server guarantees one action per candidate. Preserve anything omitted
+            // by a malformed response instead of silently losing a payment.
+            writableDatabase.rawQuery(
+                "SELECT id FROM candidates WHERE batch_id = ? AND state = 'batch_collecting'",
+                arrayOf(batchId),
+            ).use { cursor ->
+                while (cursor.moveToNext()) {
+                    writableDatabase.update(
+                        "candidates",
+                        ContentValues().apply {
+                            put("state", "auto_ready")
+                            put("ai_action", "keep")
+                            put("ai_reason", "AI 未返回该候选，已保留本地结果")
+                            put("updated_at", now)
+                        },
+                        "id = ?",
+                        arrayOf(cursor.getString(0)),
+                    )
+                    kept += 1
+                }
+            }
+            val summary = JSONObject()
+                .put("kept", kept)
+                .put("updated", updated)
+                .put("created", created)
+                .put("dropped", dropped)
+                .put("fallback", false)
+            finishBatch(batchId, "ready", summary, null, now)
+            writableDatabase.setTransactionSuccessful()
+            true
+        } finally {
+            writableDatabase.endTransaction()
+        }
+    }
+
+    @Synchronized
+    fun fallbackBatch(batchId: String, reason: String): Boolean {
+        val now = System.currentTimeMillis()
+        writableDatabase.beginTransaction()
+        return try {
+            writableDatabase.execSQL(
+                """
+                UPDATE candidates
+                SET state = CASE WHEN high_confidence = 1 THEN 'auto_ready' ELSE 'pending_confirm' END,
+                    ai_action = 'fallback', ai_reason = ?, updated_at = ?
+                WHERE batch_id = ? AND state = 'batch_collecting'
+                """.trimIndent(),
+                arrayOf<Any>(reason.take(80), now, batchId),
+            )
+            val count = readableDatabase.rawQuery(
+                "SELECT COUNT(*) FROM candidates WHERE batch_id = ? AND state IN ('auto_ready','pending_confirm')",
+                arrayOf(batchId),
+            ).use { cursor -> if (cursor.moveToFirst()) cursor.getInt(0) else 0 }
+            val summary = JSONObject()
+                .put("kept", count)
+                .put("updated", 0)
+                .put("created", 0)
+                .put("dropped", 0)
+                .put("fallback", true)
+            finishBatch(batchId, "fallback", summary, reason, now)
+            writableDatabase.setTransactionSuccessful()
+            true
+        } finally {
+            writableDatabase.endTransaction()
+        }
+    }
+
+    private fun finishBatch(
+        batchId: String,
+        state: String,
+        summary: JSONObject,
+        failureReason: String?,
+        now: Long,
+    ) {
+        writableDatabase.update(
+            "recognition_batches",
+            ContentValues().apply {
+                put("state", state)
+                put("completed_at", now)
+                put("summary_json", summary.toString())
+                if (failureReason == null) putNull("failure_reason") else put("failure_reason", failureReason.take(80))
+            },
+            "id = ?",
+            arrayOf(batchId),
+        )
+        writableDatabase.delete("batch_images", "batch_id = ?", arrayOf(batchId))
+    }
+
+    private fun applyActionFields(payload: JSONObject, action: JSONObject) {
+        action.optionalString("type")?.takeIf { it in setOf("income", "expense") }?.let {
+            payload.put("type", it)
+        }
+        action.optDouble("amount", 0.0).takeIf { it > 0 }?.let { payload.put("amount", it) }
+        action.optionalString("note")?.let {
+            payload.put("merchant", it.take(40))
+            payload.put("note", it.take(40))
+        }
+        action.optionalString("paymentMethod")?.let { payload.put("paymentMethod", it.take(40)) }
+        action.optionalString("categoryName")?.let { payload.put("categoryHint", it.take(40)) }
+        action.optLong("categoryId").takeIf { it > 0 }?.let { payload.put("categoryId", it) }
+        action.optionalInstantEpoch("occurredAt")?.let { payload.put("occurredAtEpochMs", it) }
+    }
+
+    @Synchronized
+    fun recentBatches(): List<String> {
+        cleanupBatchHistory(System.currentTimeMillis())
+        return readableDatabase.rawQuery(
+            """
+            SELECT id, state, opened_at, completed_at, summary_json, failure_reason
+            FROM recognition_batches
+            WHERE completed_at IS NOT NULL
+            ORDER BY completed_at DESC LIMIT 20
+            """.trimIndent(),
+            null,
+        ).use { cursor ->
+            buildList {
+                while (cursor.moveToNext()) {
+                    val batchId = cursor.getString(0)
+                    val items = JSONArray()
+                    readableDatabase.rawQuery(
+                        "SELECT * FROM candidates WHERE batch_id = ? ORDER BY first_seen",
+                        arrayOf(batchId),
+                    ).use { candidates ->
+                        while (candidates.moveToNext()) {
+                            val payload = decryptPayload(candidates.getString(candidates.getColumnIndexOrThrow("payload_encrypted")))
+                                ?: continue
+                            val actionIndex = candidates.getColumnIndexOrThrow("ai_action")
+                            val reasonIndex = candidates.getColumnIndexOrThrow("ai_reason")
+                            val action = if (candidates.isNull(actionIndex)) "keep" else candidates.getString(actionIndex)
+                            items.put(
+                                JSONObject()
+                                    .put("candidateId", candidates.getString(candidates.getColumnIndexOrThrow("id")))
+                                    .put("action", action)
+                                    .put("reason", if (candidates.isNull(reasonIndex)) "" else candidates.getString(reasonIndex))
+                                    .put("type", payload.optString("type"))
+                                    .put("amount", payload.optDouble("amount"))
+                                    .put("merchant", payload.optString("merchant").takeIf(String::isNotBlank))
+                                    .put("state", candidates.getString(candidates.getColumnIndexOrThrow("state")))
+                                    .put("canRestore", action == "drop" && candidates.getString(candidates.getColumnIndexOrThrow("state")) == "ai_dropped"),
+                            )
+                        }
+                    }
+                    add(
+                        JSONObject()
+                            .put("id", batchId)
+                            .put("state", cursor.getString(1))
+                            .put("openedAt", cursor.getLong(2))
+                            .put("completedAt", if (cursor.isNull(3)) JSONObject.NULL else cursor.getLong(3))
+                            .put("summary", cursor.getString(4)?.let(::JSONObject) ?: JSONObject())
+                            .put("failureReason", if (cursor.isNull(5)) JSONObject.NULL else cursor.getString(5))
+                            .put("items", items)
+                            .toString(),
+                    )
+                }
+            }
+        }
+    }
+
+    @Synchronized
+    fun restoreDropped(candidateId: String): StoredCandidate? {
+        val changed = writableDatabase.update(
+            "candidates",
+            ContentValues().apply {
+                put("state", "auto_ready")
+                put("ai_action", "restored")
+                put("ai_reason", "用户恢复 AI 剔除项")
+                put("updated_at", System.currentTimeMillis())
+            },
+            "id = ? AND state = 'ai_dropped'",
+            arrayOf(candidateId),
+        )
+        return if (changed == 1) loadById(candidateId) else null
+    }
+
+    private fun cleanupBatchHistory(now: Long) {
+        val cutoff = now - BATCH_HISTORY_MS
+        writableDatabase.delete("batch_images", "batch_id IN (SELECT id FROM recognition_batches WHERE completed_at < ?)", arrayOf(cutoff.toString()))
+        writableDatabase.execSQL(
+            """
+            UPDATE candidates
+            SET state = CASE WHEN state = 'ai_dropped' THEN 'expired' ELSE state END,
+                batch_id = NULL, ai_action = NULL, ai_reason = NULL,
+                original_payload_encrypted = NULL
+            WHERE batch_id IN (
+              SELECT id FROM recognition_batches WHERE completed_at < ?
+            )
+            """.trimIndent(),
+            arrayOf(cutoff),
+        )
+        writableDatabase.delete("recognition_batches", "completed_at < ?", arrayOf(cutoff.toString()))
+    }
+
+    private fun JSONObject.optionalString(name: String): String? =
+        if (!has(name) || isNull(name)) null else optString(name).trim().takeIf(String::isNotBlank)
+
+    private fun JSONObject.optionalInstantEpoch(name: String): Long? =
+        optionalString(name)?.let { value -> runCatching { Instant.parse(value).toEpochMilli() }.getOrNull() }
+
    @Synchronized
    fun finalizeCandidate(id: String): StoredCandidate? {
        val now = System.currentTimeMillis()
@@ -277,6 +906,9 @@ class RecognitionStore(context: Context) :
            ).use { cursor ->
                if (cursor.moveToFirst()) return row(cursor)
            }
+            // A new accessibility/OCR flow is a new payment, even when amount and
+            // counterparty are identical to another transaction in the same window.
+            return null
        }
        val since = signal.occurredAtEpochMs - NO_ORDER_WINDOW_MS
        val until = signal.occurredAtEpochMs + NO_ORDER_WINDOW_MS
@@ -327,7 +959,17 @@ class RecognitionStore(context: Context) :
            .put("clientRequestId", cursor.getString(cursor.getColumnIndexOrThrow("client_request_id")))
            .put("state", cursor.getString(cursor.getColumnIndexOrThrow("state")))
            .put("confidence", if (cursor.getInt(cursor.getColumnIndexOrThrow("high_confidence")) == 1) "auto" else "confirm")
-            .put("source", sourceFromMask(cursor.getInt(cursor.getColumnIndexOrThrow("channel_mask"))))
+            .put(
+                "source",
+                payload.optString("sourceOverride").takeIf(String::isNotBlank)
+                    ?: sourceFromMask(cursor.getInt(cursor.getColumnIndexOrThrow("channel_mask"))),
+            )
+        val batchIndex = cursor.getColumnIndex("batch_id")
+        val actionIndex = cursor.getColumnIndex("ai_action")
+        val reasonIndex = cursor.getColumnIndex("ai_reason")
+        if (batchIndex >= 0 && !cursor.isNull(batchIndex)) payload.put("batchId", cursor.getString(batchIndex))
+        if (actionIndex >= 0 && !cursor.isNull(actionIndex)) payload.put("aiAction", cursor.getString(actionIndex))
+        if (reasonIndex >= 0 && !cursor.isNull(reasonIndex)) payload.put("aiReason", cursor.getString(reasonIndex))
        val txIndex = cursor.getColumnIndexOrThrow("transaction_id")
        val txId = if (cursor.isNull(txIndex)) null else cursor.getLong(txIndex)
        if (txId != null) payload.put("transactionId", txId)
@@ -337,6 +979,9 @@ class RecognitionStore(context: Context) :
            state = payload.getString("state"),
            transactionId = txId,
            json = payload.toString(),
+            batchId = cursor.getColumnIndex("batch_id").takeIf { it >= 0 }?.let { index ->
+                if (cursor.isNull(index)) null else cursor.getString(index)
+            },
        )
    }

@@ -352,6 +997,9 @@ class RecognitionStore(context: Context) :
            updatedAt = cursor.getLong(cursor.getColumnIndexOrThrow("updated_at")),
            resultFingerprint = payload.optString("resultFingerprint").takeIf(String::isNotBlank),
            payload = payload,
+            batchId = cursor.getColumnIndex("batch_id").takeIf { it >= 0 }?.let { index ->
+                if (cursor.isNull(index)) null else cursor.getString(index)
+            },
        )
    }

@@ -405,7 +1053,7 @@ class RecognitionStore(context: Context) :
    private fun expireOld(now: Long) {
        writableDatabase.execSQL(
            "UPDATE candidates SET state = 'expired', updated_at = ? " +
-                "WHERE state NOT IN ('imported','undone','expired') AND first_seen < ?",
+                "WHERE state NOT IN ('imported','undone','expired','ai_dropped') AND first_seen < ?",
            arrayOf(now, now - EXPIRE_MS),
        )
        writableDatabase.delete("evidence", "created_at < ?", arrayOf((now - EXPIRE_MS).toString()))
@@ -435,6 +1083,7 @@ class RecognitionStore(context: Context) :
        val updatedAt: Long,
        val resultFingerprint: String?,
        val payload: JSONObject,
+        val batchId: String?,
    )

    companion object {
@@ -443,6 +1092,11 @@ class RecognitionStore(context: Context) :
        private const val NO_ORDER_WINDOW_MS = 90_000L
        private const val SAME_CHANNEL_DEBOUNCE_MS = 10_000L
        private const val EXPIRE_MS = 7L * 24L * 60L * 60L * 1000L
+        private const val BATCH_IDLE_MS = 30_000L
+        private const val BATCH_HARD_LIMIT_MS = 120_000L
+        private const val BATCH_HISTORY_MS = 7L * 24L * 60L * 60L * 1000L
+        private const val MAX_BATCH_ITEMS = 10
+        private const val MAX_BATCH_IMAGE_BYTES = 1024 * 1024

        internal fun clientRequestIdFor(signal: PaymentSignal): String {
            val basis = when {
@@ -455,5 +1109,10 @@ class RecognitionStore(context: Context) :
            }
            return "recognition-${PaymentParser.sha256(basis).take(52)}"
        }
+
+        internal fun flowStrongKeyFor(signal: PaymentSignal): String? =
+            signal.flowSessionId?.takeIf { it.isNotBlank() }?.let {
+                PaymentParser.sha256("${signal.packageName}|flow|$it")
+            }
    }
 }
@@ -33,6 +33,7 @@ class ScreenshotAccessibilityService : AccessibilityService() {
    private var ocrInProgress = false
    private var visualOperationId: String? = null
    private var visualTimeout: Runnable? = null
+    private var batchCaptureTimeout: Runnable? = null

    private data class CollectedPage(val text: String, val nodeCount: Int)

@@ -53,6 +54,7 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        var resultPageHash: String? = null,
        var resultFingerprint: String? = null,
        var completed: Boolean = false,
+        var resultSurfaceExited: Boolean = false,
        var retryCount: Int = 0,
        var probeCount: Int = 0,
    )
@@ -146,6 +148,9 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        val resultSurface = status.strength != PaymentStatusStrength.NONE ||
            (existingFlow?.kind == "red_packet_send" &&
                PaymentParser.hasRedPacketSentSurface(combined))
+        if (existingFlow?.completed == true && !resultSurface && combined.isNotBlank()) {
+            existingFlow.resultSurfaceExited = true
+        }
        if (existingFlow?.completed == true && resultSurface) {
            val currentKind = if (existingFlow.kind == "red_packet_send" &&
                PaymentParser.hasRedPacketSentSurface(combined)
@@ -160,7 +165,9 @@ class ScreenshotAccessibilityService : AccessibilityService() {
            val sameOutgoingResult =
                currentKind in setOf("payment", "transfer") &&
                    existingFlow.kind in setOf("payment", "transfer")
-            if (currentKind == existingFlow.kind || sameOutgoingResult) {
+            if ((currentKind == existingFlow.kind || sameOutgoingResult) &&
+                shouldSuppressCompletedResult(existingFlow.resultSurfaceExited)
+            ) {
                RecognitionDiagnostics.record(
                    this,
                    recognizedPackage,
@@ -282,6 +289,8 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        pendingVisualCapture = null
        visualTimeout?.let(handler::removeCallbacks)
        visualTimeout = null
+        batchCaptureTimeout?.let(handler::removeCallbacks)
+        batchCaptureTimeout = null
        visualOperationId = null
        captureInProgress = false
        ocrInProgress = false
@@ -392,20 +401,43 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        nodeCount: Int,
        stage: String,
        recordDiagnostic: Boolean = true,
+        evidenceImage: ByteArray? = null,
    ) {
-        if (flow.completed) return
+        if (flow.completed) {
+            evidenceImage?.fill(0)
+            return
+        }
        flow.completed = true
        flow.resultFingerprint = signal.resultFingerprint
-        RecognitionCoordinator.get(this).submit(signal)
+        val coordinator = RecognitionCoordinator.get(this)
+        val settings = RecognitionSettings.snapshot(this)
+        val batchEnabled = settings.aiScreenshot && settings.aiAllowed && settings.hasAccount
+        if (evidenceImage != null) {
+            coordinator.submit(signal, evidenceImage)
+        } else if (batchEnabled) {
+            coordinator.submit(signal) { submission ->
+                if (submission.batchId != null) {
+                    captureBatchEvidence(signal, submission)
+                }
+            }
+        } else {
+            coordinator.submit(signal)
+        }
        if (recordDiagnostic) {
            RecognitionDiagnostics.record(
                this,
                signal.packageName,
                stage = stage,
-                result = if (signal.evidenceConfidence == "high") "auto_ready" else "confirm",
+                result = if (batchEnabled) {
+                    "batched"
+                } else if (signal.evidenceConfidence == "high") {
+                    "auto_ready"
+                } else {
+                    "confirm"
+                },
                nodeCount = nodeCount,
                amountCandidates = 1,
-                reason = "success",
+                reason = if (batchEnabled) "queued_for_ai" else "success",
                expectedAmountMatched = flow.expectedAmountCents?.let {
                    it == signal.amountCents
                },
@@ -695,12 +727,21 @@ class ScreenshotAccessibilityService : AccessibilityService() {
            )
            val signal = outcome.signal
            if (signal != null) {
+                val evidence = if (RecognitionSettings.snapshot(this).let {
+                        it.aiScreenshot && it.aiAllowed && it.hasAccount
+                    }
+                ) {
+                    runCatching { bitmapToBatchBytes(bitmap) }.getOrNull()
+                } else {
+                    null
+                }
                submitOnce(
                    signal,
                    flow,
                    nodeCount,
                    "local_ocr",
                    recordDiagnostic = false,
+                    evidenceImage = evidence,
                )
                return
            }
@@ -725,8 +766,13 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        val settings = RecognitionSettings.snapshot(this)
        if (!settings.aiScreenshot || !settings.aiAllowed || !settings.hasAccount) return
        runCatching {
-            val bytes = bitmapToBytes(bitmap)
-            BackgroundAiRecognizer.analyze(this, flow.packageName, bytes, flow.id)
+            val bytes = bitmapToBatchBytes(bitmap)
+            RecognitionCoordinator.get(this).submitEvidenceOnly(
+                flow.packageName,
+                flow.id,
+                System.currentTimeMillis(),
+                bytes,
+            )
        }.onFailure {
            RecognitionDiagnostics.record(
                this,
@@ -738,6 +784,71 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        }
    }

+    private fun captureBatchEvidence(signal: PaymentSignal, submission: StoredSubmission) {
+        if (Build.VERSION.SDK_INT < Build.VERSION_CODES.R || captureInProgress) return
+        captureInProgress = true
+        var finished = false
+        fun finish(): Boolean {
+            if (finished) return false
+            finished = true
+            batchCaptureTimeout?.let(handler::removeCallbacks)
+            batchCaptureTimeout = null
+            captureInProgress = false
+            return true
+        }
+        batchCaptureTimeout = Runnable {
+            if (finish()) {
+                RecognitionDiagnostics.record(
+                    this,
+                    signal.packageName,
+                    stage = "ai_batch_capture",
+                    result = "failed",
+                    reason = "capture_timeout",
+                )
+            }
+        }.also { handler.postDelayed(it, CAPTURE_CALLBACK_TIMEOUT_MS) }
+        val callback = object : TakeScreenshotCallback {
+            override fun onSuccess(screenshot: ScreenshotResult) {
+                if (!finish()) {
+                    screenshot.hardwareBuffer.close()
+                    return
+                }
+                var bitmap: Bitmap? = null
+                runCatching {
+                    bitmap = copyBitmap(screenshot)
+                    val bytes = bitmapToBatchBytes(requireNotNull(bitmap))
+                    RecognitionCoordinator.get(this@ScreenshotAccessibilityService)
+                        .attachBatchImage(submission, signal, bytes)
+                }.onFailure {
+                    RecognitionDiagnostics.record(
+                        this@ScreenshotAccessibilityService,
+                        signal.packageName,
+                        stage = "ai_batch_capture",
+                        result = "failed",
+                        reason = "image_encode_failed",
+                    )
+                }
+                bitmap?.recycle()
+            }
+
+            override fun onFailure(errorCode: Int) {
+                if (!finish()) return
+                RecognitionDiagnostics.record(
+                    this@ScreenshotAccessibilityService,
+                    signal.packageName,
+                    stage = "ai_batch_capture",
+                    result = "failed",
+                    reason = screenshotDiagnosticReason(errorCode),
+                )
+            }
+        }
+        runCatching {
+            takeScreenshot(Display.DEFAULT_DISPLAY, mainExecutor, callback)
+        }.onFailure {
+            finish()
+        }
+    }
+
    private fun handleVisualFailure(flow: PaymentFlow, nodeCount: Int, reason: String) {
        RecognitionDiagnostics.record(
            this,
@@ -918,13 +1029,43 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        }
    }

-    private fun bitmapToBytes(bitmap: Bitmap): ByteArray =
-        ByteArrayOutputStream().use { output ->
-            check(bitmap.compress(Bitmap.CompressFormat.PNG, 92, output)) {
-                "无法编码截屏"
-            }
-            output.toByteArray()
+    private fun bitmapToBatchBytes(bitmap: Bitmap): ByteArray {
+        val longest = maxOf(bitmap.width, bitmap.height)
+        val scaled = if (longest > BATCH_IMAGE_MAX_EDGE) {
+            val ratio = BATCH_IMAGE_MAX_EDGE.toDouble() / longest
+            Bitmap.createScaledBitmap(
+                bitmap,
+                (bitmap.width * ratio).toInt().coerceAtLeast(1),
+                (bitmap.height * ratio).toInt().coerceAtLeast(1),
+                true,
+            )
+        } else {
+            bitmap
        }
+        return try {
+            var quality = 82
+            var bytes = ByteArray(0)
+            try {
+                do {
+                    bytes.fill(0)
+                    bytes = ByteArrayOutputStream().use { output ->
+                        check(scaled.compress(Bitmap.CompressFormat.JPEG, quality, output)) {
+                            "无法编码批次截图"
+                        }
+                        output.toByteArray()
+                    }
+                    quality -= 10
+                } while (bytes.size > BATCH_IMAGE_MAX_BYTES && quality >= 52)
+                check(bytes.size <= BATCH_IMAGE_MAX_BYTES) { "批次截图压缩后仍然过大" }
+                bytes
+            } catch (error: Exception) {
+                bytes.fill(0)
+                throw error
+            }
+        } finally {
+            if (scaled !== bitmap) scaled.recycle()
+        }
+    }

    private fun copyBitmap(screenshot: ScreenshotResult): Bitmap {
        val buffer = screenshot.hardwareBuffer
@@ -976,6 +1117,8 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        private const val MAX_TREE_NODES = 160
        private const val MAX_CHILDREN_PER_NODE = 40
        private const val MAX_TEXT_CHARS = 8_000
+        private const val BATCH_IMAGE_MAX_EDGE = 1280
+        private const val BATCH_IMAGE_MAX_BYTES = 900 * 1024
        private val MAJOR_WINDOW_EVENTS = setOf(
            AccessibilityEvent.TYPE_WINDOW_STATE_CHANGED,
            AccessibilityEvent.TYPE_WINDOWS_CHANGED,
@@ -991,6 +1134,9 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        @Volatile
        private var activeInstance: ScreenshotAccessibilityService? = null

+        internal fun shouldSuppressCompletedResult(resultSurfaceExited: Boolean): Boolean =
+            !resultSurfaceExited
+
        @Volatile
        var isConnected = false
            private set
@@ -1010,4 +1156,3 @@ class ScreenshotAccessibilityService : AccessibilityService() {
        }
    }
 }
-
@@ -301,6 +301,47 @@ class PaymentParserTest {
        )
    }

+    @Test
+    fun completedResultStaysDeduplicatedUntilTheSurfaceIsExited() {
+        assertTrue(ScreenshotAccessibilityService.shouldSuppressCompletedResult(false))
+        assertFalse(ScreenshotAccessibilityService.shouldSuppressCompletedResult(true))
+    }
+
+    @Test
+    fun consecutiveIdenticalTransfersKeepDistinctFlowIdentities() {
+        val first = paymentSignal(
+            channel = "accessibility",
+            sourceEventId = "a:wechat:transfer-1",
+            flowSessionId = "transfer-1",
+        )
+        val second = paymentSignal(
+            channel = "accessibility",
+            sourceEventId = "a:wechat:transfer-2",
+            flowSessionId = "transfer-2",
+        )
+        val third = paymentSignal(
+            channel = "accessibility",
+            sourceEventId = "a:wechat:transfer-3",
+            flowSessionId = "transfer-3",
+        ).copy(amountCents = 3_000L)
+
+        assertNotEquals(
+            RecognitionStore.flowStrongKeyFor(first),
+            RecognitionStore.flowStrongKeyFor(second),
+        )
+        assertNotEquals(
+            RecognitionStore.flowStrongKeyFor(second),
+            RecognitionStore.flowStrongKeyFor(third),
+        )
+        assertEquals(
+            3,
+            listOf(first, second, third)
+                .map(RecognitionStore::flowStrongKeyFor)
+                .toSet()
+                .size,
+        )
+    }
+
    private fun paymentSignal(
        channel: String,
        sourceEventId: String,