profile: add adaptive daily request control

2026-09-15 02:56:10 +08:00 · 2026-08-06 16:00:04 +08:00
parent 3c69bdeff4
commit c7878030f9
6 changed files with 693 additions and 27 deletions
@@ -107,13 +107,30 @@ object ProfileDailyMaintenance {

        JChatGPT.logger.info("每日群画像推进开始: groups=${groupIds.size} maxPendingAgeDays=$maxPendingAgeDays")
        val runToken = UserProfileAnalysisService.newRunToken()
+        val requestController = ProfileDailyRequestController(
+            maxConcurrentRequests = PluginConfig.profileMaxConcurrentRequests,
+            maxAttempts = PluginConfig.profileRetryMax.coerceIn(0, 3) + 1,
+        )
        val outcomes = coroutineScope {
            groupIds.map { groupId ->
-                async { analyzeGroup(groupId, runToken) }
+                async { analyzeGroup(groupId, runToken, requestController) }
            }.awaitAll()
        }
+        val requestStats = requestController.snapshot()

        val reports = outcomes.mapNotNull(DailyGroupOutcome::report)
+        outcomes.firstNotNullOfOrNull(DailyGroupOutcome::cause)?.let { cause ->
+            JChatGPT.logger.error("每日群画像推进存在失败群，本轮仅输出一次代表性异常", cause)
+        }
+        if (requestStats.countedFailures > 0) {
+            val message = "每日群画像请求控制: attempts=${requestStats.totalAttempts} " +
+                    "failures=${requestStats.countedFailures} probes=${requestStats.probeAttempts} " +
+                    "incidents=${requestStats.pauseIncidents} recovered=${requestStats.recoveredIncidents} " +
+                    "exhausted=${requestStats.exhaustedTasks} " +
+                    "peak=${requestStats.peakActive}/${requestStats.peakLimit} " +
+                    "stopped=${requestStats.stopped}"
+            if (requestStats.stopped) JChatGPT.logger.warning(message) else JChatGPT.logger.info(message)
+        }
        JChatGPT.logger.info(
            "每日群画像推进完成: selected=${groupIds.size} " +
                    "success=${outcomes.count { it.status == DailyGroupStatus.SUCCESS }} " +
@@ -132,12 +149,14 @@ object ProfileDailyMaintenance {
    private suspend fun analyzeGroup(
        groupId: Long,
        runToken: ProfileAnalysisRunToken,
+        requestController: ProfileDailyRequestController,
    ): DailyGroupOutcome {
        return try {
-            val report = UserProfileAnalysisService.analyzeGroup(
+            val report = UserProfileAnalysisService.analyzeGroupControlled(
                groupId = groupId,
                maxBatches = Int.MAX_VALUE,
                runToken = runToken,
+                requestController = requestController,
            ) { progress ->
                JChatGPT.logger.info(
                    "PROFILE_DAILY_BATCH group=$groupId batch=${progress.batchIndex} " +
@@ -155,17 +174,19 @@ object ProfileDailyMaintenance {
                else -> DailyGroupStatus.SUCCESS
            }
            DailyGroupOutcome(status, report.takeUnless { report.alreadyRunning || report.botId == null })
+        } catch (cause: ProfileDailyRunStoppedException) {
+            DailyGroupOutcome(DailyGroupStatus.STOPPED)
        } catch (cause: CancellationException) {
            throw cause
        } catch (cause: Exception) {
-            JChatGPT.logger.error("群 $groupId 每日画像推进失败", cause)
-            DailyGroupOutcome(DailyGroupStatus.FAILED)
+            DailyGroupOutcome(DailyGroupStatus.FAILED, cause = cause)
        }
    }

    private data class DailyGroupOutcome(
        val status: DailyGroupStatus,
        val report: GroupProfileAnalysisReport? = null,
+        val cause: Throwable? = null,
    )

    private enum class DailyGroupStatus {
@@ -0,0 +1,304 @@
+package top.jie65535.mirai.profile
+
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import top.jie65535.mirai.llm.normalizeMaxConcurrentRequests
+
+/**
+ * Run-scoped admission controller for the online daily group maintenance job.
+ *
+ * The shared ModelService semaphore remains the hard upper bound. This gate
+ * only controls this one bulk run: it ramps healthy admission from one request,
+ * pauses fresh work after a failure, and keeps recovery retries single-flight.
+ */
+internal class ProfileDailyRequestController(
+    maxConcurrentRequests: Int,
+    private val maxAttempts: Int,
+    private val growthFactor: Int = 4,
+    private val maxConsecutiveExhaustedTasks: Int = 3,
+) {
+    init {
+        require(maxAttempts > 0) { "maxAttempts must be positive" }
+        require(growthFactor >= 2) { "growthFactor must be at least 2" }
+        require(maxConsecutiveExhaustedTasks > 0) {
+            "maxConsecutiveExhaustedTasks must be positive"
+        }
+    }
+
+    private val maxConcurrency = normalizeMaxConcurrentRequests(maxConcurrentRequests)
+    private val mutex = Mutex()
+    private val waiters = ArrayDeque<Waiter>()
+    private val active = LinkedHashMap<Long, Permit>()
+    private val heldFailures = LinkedHashSet<ProfileDailyRequestKey>()
+    private val exhaustedTasks = LinkedHashSet<ProfileDailyRequestKey>()
+
+    private var nextPermitId = 1L
+    private var admissionLimit = 1
+    private var successesAtLimit = 0
+    private var paused = false
+    private var incidentOpen = false
+    private var stopped = false
+    private var consecutiveExhaustedTasks = 0
+    private var freshProbeSuccesses = 0
+
+    private var totalAttempts = 0
+    private var countedFailures = 0
+    private var probeAttempts = 0
+    private var pauseIncidents = 0
+    private var recoveredIncidents = 0
+    private var peakActive = 0
+    private var peakLimit = 1
+    private var maxParallelProbes = 0
+
+    /**
+     * Admits and records one model attempt. A rejected/safety response can be
+     * excluded from the global failure circuit without affecting the gate.
+     */
+    suspend fun <T> execute(
+        key: ProfileDailyRequestKey,
+        attempt: Int,
+        countFailure: (Throwable) -> Boolean = { true },
+        block: suspend () -> T,
+    ): T {
+        require(attempt in 1..maxAttempts) { "attempt must be within 1..$maxAttempts" }
+        val permit = awaitPermit(key, attempt)
+        return try {
+            val result = block()
+            onSuccess(permit)
+            result
+        } catch (cause: CancellationException) {
+            onCancellation(permit)
+            throw cause
+        } catch (cause: Throwable) {
+            if (countFailure(cause)) onFailure(permit, attempt)
+            else release(permit)
+            throw cause
+        }
+    }
+
+    suspend fun snapshot(): ProfileDailyRequestControllerStats = mutex.withLock {
+        ProfileDailyRequestControllerStats(
+            totalAttempts = totalAttempts,
+            countedFailures = countedFailures,
+            probeAttempts = probeAttempts,
+            pauseIncidents = pauseIncidents,
+            recoveredIncidents = recoveredIncidents,
+            exhaustedTasks = exhaustedTasks.size,
+            stopped = stopped,
+            peakActive = peakActive,
+            peakLimit = peakLimit,
+            finalLimit = admissionLimit,
+            maxParallelProbes = maxParallelProbes,
+        )
+    }
+
+    private suspend fun awaitPermit(
+        key: ProfileDailyRequestKey,
+        attempt: Int,
+    ): Permit {
+        val waiter = Waiter(key, attempt)
+        mutex.withLock {
+            if (stopped) {
+                waiter.deferred.completeExceptionally(ProfileDailyRunStoppedException)
+            } else {
+                waiters.addLast(waiter)
+                pumpLocked()
+            }
+        }
+        return try {
+            waiter.deferred.await()
+        } catch (cause: CancellationException) {
+            mutex.withLock {
+                waiters.remove(waiter)
+                waiter.permit?.let { permit -> active.remove(permit.id) }
+                waiter.deferred.cancel(cause)
+                pumpLocked()
+            }
+            throw cause
+        }
+    }
+
+    private suspend fun onSuccess(permit: Permit) {
+        mutex.withLock {
+            active.remove(permit.id)
+            if (permit.kind == PermitKind.RECOVERY) {
+                heldFailures.remove(permit.key)
+                if (paused) resumeAfterRecoverySuccessLocked()
+            } else if (permit.kind == PermitKind.FRESH_PROBE) {
+                freshProbeSuccesses++
+            }
+            consecutiveExhaustedTasks = 0
+            successesAtLimit++
+            if (admissionLimit < maxConcurrency && successesAtLimit >= admissionLimit) {
+                admissionLimit = minOf(maxConcurrency, admissionLimit * growthFactor)
+                successesAtLimit = 0
+                peakLimit = maxOf(peakLimit, admissionLimit)
+            }
+            if (paused && heldFailures.isEmpty() && freshProbeSuccesses >= FRESH_PROBE_SUCCESS_TARGET) {
+                resumeAfterFreshProbesLocked()
+            }
+            pumpLocked()
+        }
+    }
+
+    private suspend fun onFailure(permit: Permit, attempt: Int) {
+        mutex.withLock {
+            active.remove(permit.id)
+            countedFailures++
+            if (stopped) return@withLock
+
+            if (!incidentOpen) {
+                incidentOpen = true
+                paused = true
+                pauseIncidents++
+            }
+            heldFailures.add(permit.key)
+            if (attempt >= maxAttempts) {
+                heldFailures.remove(permit.key)
+                if (exhaustedTasks.add(permit.key)) consecutiveExhaustedTasks++
+                if (consecutiveExhaustedTasks >= maxConsecutiveExhaustedTasks) {
+                    stopped = true
+                    failWaitingLocked()
+                }
+            }
+            pumpLocked()
+        }
+    }
+
+    private suspend fun onCancellation(permit: Permit) {
+        mutex.withLock {
+            active.remove(permit.id)
+            pumpLocked()
+        }
+    }
+
+    private suspend fun release(permit: Permit) {
+        mutex.withLock {
+            active.remove(permit.id)
+            pumpLocked()
+        }
+    }
+
+    private fun resumeAfterRecoverySuccessLocked() {
+        paused = false
+        if (incidentOpen) {
+            recoveredIncidents++
+            incidentOpen = false
+        }
+        freshProbeSuccesses = 0
+    }
+
+    private fun resumeAfterFreshProbesLocked() {
+        paused = false
+        if (incidentOpen) {
+            recoveredIncidents++
+            incidentOpen = false
+        }
+        freshProbeSuccesses = 0
+    }
+
+    private fun pumpLocked() {
+        removeInactiveWaitersLocked()
+        if (stopped) return
+
+        while (active.size < admissionLimit) {
+            val waiter = when {
+                !paused -> findNormalOrRecoveryWaiterLocked()
+                heldFailures.isNotEmpty() -> {
+                    if (active.values.any { it.kind == PermitKind.RECOVERY }) null
+                    else waiters.firstOrNull { it.attempt > 1 && heldFailures.contains(it.key) }
+                }
+                freshProbeSuccesses < FRESH_PROBE_SUCCESS_TARGET -> {
+                    waiters.firstOrNull { it.attempt == 1 }
+                }
+                else -> null
+            } ?: break
+
+            waiters.remove(waiter)
+            val kind = when {
+                waiter.attempt > 1 && heldFailures.contains(waiter.key) -> PermitKind.RECOVERY
+                paused -> PermitKind.FRESH_PROBE
+                else -> PermitKind.NORMAL
+            }
+            val permit = Permit(nextPermitId++, waiter.key, kind)
+            waiter.permit = permit
+            active[permit.id] = permit
+            totalAttempts++
+            if (kind != PermitKind.NORMAL) probeAttempts++
+            peakActive = maxOf(peakActive, active.size)
+            maxParallelProbes = maxOf(
+                maxParallelProbes,
+                active.values.count { it.kind != PermitKind.NORMAL },
+            )
+            waiter.deferred.complete(permit)
+
+            if (kind == PermitKind.RECOVERY) break
+            if (paused) break
+        }
+    }
+
+    private fun findNormalOrRecoveryWaiterLocked(): Waiter? {
+        if (heldFailures.isNotEmpty() && active.values.none { it.kind == PermitKind.RECOVERY }) {
+            return waiters.firstOrNull { it.attempt > 1 && heldFailures.contains(it.key) }
+        }
+        return waiters.firstOrNull { it.attempt == 1 }
+    }
+
+    private fun removeInactiveWaitersLocked() {
+        waiters.removeAll { !it.deferred.isActive }
+    }
+
+    private fun failWaitingLocked() {
+        waiters.forEach { it.deferred.completeExceptionally(ProfileDailyRunStoppedException) }
+        waiters.clear()
+    }
+
+    private data class Waiter(
+        val key: ProfileDailyRequestKey,
+        val attempt: Int,
+        val deferred: CompletableDeferred<Permit> = CompletableDeferred(),
+        var permit: Permit? = null,
+    )
+
+    private data class Permit(
+        val id: Long,
+        val key: ProfileDailyRequestKey,
+        val kind: PermitKind,
+    )
+
+    private enum class PermitKind {
+        NORMAL,
+        RECOVERY,
+        FRESH_PROBE,
+    }
+
+    companion object {
+        private const val FRESH_PROBE_SUCCESS_TARGET = 2
+    }
+}
+
+internal data class ProfileDailyRequestKey(
+    val groupId: Long,
+    val startTime: Int,
+    val endTime: Int,
+)
+
+internal data class ProfileDailyRequestControllerStats(
+    val totalAttempts: Int,
+    val countedFailures: Int,
+    val probeAttempts: Int,
+    val pauseIncidents: Int,
+    val recoveredIncidents: Int,
+    val exhaustedTasks: Int,
+    val stopped: Boolean,
+    val peakActive: Int,
+    val peakLimit: Int,
+    val finalLimit: Int,
+    val maxParallelProbes: Int,
+)
+
+internal object ProfileDailyRunStoppedException : RuntimeException(
+    "每日群画像推进已因连续模型请求失败而停止",
+)
@@ -170,6 +170,34 @@ object UserProfileAnalysisService {
        maxBatches: Int,
        runToken: ProfileAnalysisRunToken = newRunToken(),
        onProgress: suspend (GroupProfileAnalysisProgress) -> Unit = {},
+    ): GroupProfileAnalysisReport = analyzeGroupInternal(
+        groupId = groupId,
+        maxBatches = maxBatches,
+        runToken = runToken,
+        requestController = null,
+        onProgress = onProgress,
+    )
+
+    internal suspend fun analyzeGroupControlled(
+        groupId: Long,
+        maxBatches: Int,
+        runToken: ProfileAnalysisRunToken,
+        requestController: ProfileDailyRequestController,
+        onProgress: suspend (GroupProfileAnalysisProgress) -> Unit = {},
+    ): GroupProfileAnalysisReport = analyzeGroupInternal(
+        groupId = groupId,
+        maxBatches = maxBatches,
+        runToken = runToken,
+        requestController = requestController,
+        onProgress = onProgress,
+    )
+
+    private suspend fun analyzeGroupInternal(
+        groupId: Long,
+        maxBatches: Int,
+        runToken: ProfileAnalysisRunToken,
+        requestController: ProfileDailyRequestController?,
+        onProgress: suspend (GroupProfileAnalysisProgress) -> Unit,
    ): GroupProfileAnalysisReport {
        require(groupId > 0) { "groupId 必须是正数" }
        require(maxBatches > 0) { "maxBatches 必须是正数" }
@@ -194,7 +222,7 @@ object UserProfileAnalysisService {
        }

        try {
-            return analyzeGroupExclusive(groupId, maxBatches, runToken, onProgress)
+            return analyzeGroupExclusive(groupId, maxBatches, runToken, requestController, onProgress)
        } finally {
            runningGroups.remove(groupId)
        }
@@ -306,6 +334,7 @@ object UserProfileAnalysisService {
        groupId: Long,
        maxBatches: Int,
        runToken: ProfileAnalysisRunToken,
+        requestController: ProfileDailyRequestController?,
        onProgress: suspend (GroupProfileAnalysisProgress) -> Unit,
    ): GroupProfileAnalysisReport {
        val reader = withContext(Dispatchers.IO) { ProfileHistoryReader(resolveHistoryFile()) }
@@ -367,6 +396,7 @@ object UserProfileAnalysisService {
                    summaryMaxLength = PluginConfig.profileSummaryMaxLength,
                    onRetryFailure = { message, cause -> JChatGPT.logger.warning(message, cause) },
                    onCommittedOperations = ProfileOperationLogger::log,
+                    requestController = requestController,
                )
            } catch (cause: ModelSafetyRejectionException) {
                JChatGPT.logger.warning(
@@ -461,6 +491,7 @@ object UserProfileAnalysisService {
        summaryMaxLength: Int,
        onRetryFailure: (String, Throwable) -> Unit = { _, _ -> },
        onCommittedOperations: (String, Collection<ProfileReduction>) -> Unit = { _, _ -> },
+        requestController: ProfileDailyRequestController? = null,
    ): ConversationProfileAnalysisReport? {
        check(UserProfileStore.isAvailable) { "用户画像数据库不可用" }
        val eligibleUserIds = batch.authoredTextCharsByUser
@@ -486,6 +517,7 @@ object UserProfileAnalysisService {
            retryMax = retryMax,
            summaryMaxLength = summaryMaxLength,
            onRetryFailure = onRetryFailure,
+            requestController = requestController,
        )
        val totalUsage = result.usage
        val commitOutcome = userLocks.withUserLocks(eligibleUserIds) {
@@ -569,36 +601,57 @@ object UserProfileAnalysisService {
        retryMax: Int,
        summaryMaxLength: Int,
        onRetryFailure: (String, Throwable) -> Unit,
+        requestController: ProfileDailyRequestController?,
    ): Pair<ConversationProfileModelResult, List<ProfileReduction>> {
        val attempts = retryMax.coerceIn(0, 3) + 1
        val retryBackoff = RetryBackoff.fromConfig()
+        val requestKey = ProfileDailyRequestKey(
+            groupId = batch.groupId,
+            startTime = batch.startTime,
+            endTime = batch.endTime,
+        )
        var lastFailure: Throwable? = null
        repeat(attempts) { attempt ->
            try {
-                val result = model.analyzeConversation(
-                    profiles,
-                    batch,
-                    eligibleUserIds,
-                    supportStatsByUserId,
-                )
-                val reductions = ConversationProfileReducer.reduce(
-                    profiles = profiles,
-                    batch = batch,
-                    eligibleUserIds = eligibleUserIds,
-                    response = result.response,
-                    model = model.modelName,
-                    promptVersion = ProfilePromptStore.PROMPT_VERSION,
-                    summaryMaxLength = summaryMaxLength.coerceAtLeast(100),
-                )
-                return result to reductions
+                val analyzeAttempt: suspend () -> Pair<ConversationProfileModelResult, List<ProfileReduction>> = {
+                    val result = model.analyzeConversation(
+                        profiles,
+                        batch,
+                        eligibleUserIds,
+                        supportStatsByUserId,
+                    )
+                    val reductions = ConversationProfileReducer.reduce(
+                        profiles = profiles,
+                        batch = batch,
+                        eligibleUserIds = eligibleUserIds,
+                        response = result.response,
+                        model = model.modelName,
+                        promptVersion = ProfilePromptStore.PROMPT_VERSION,
+                        summaryMaxLength = summaryMaxLength.coerceAtLeast(100),
+                    )
+                    result to reductions
+                }
+                return if (requestController == null) {
+                    analyzeAttempt()
+                } else {
+                    requestController.execute(
+                        key = requestKey,
+                        attempt = attempt + 1,
+                        countFailure = { cause -> cause !is ModelRequestRejectedException },
+                        block = analyzeAttempt,
+                    )
+                }
            } catch (cause: Exception) {
+                if (cause is ProfileDailyRunStoppedException) throw cause
                if (cause is CancellationException) throw cause
                if (cause is ModelRequestRejectedException) {
-                    onRetryFailure(
-                        "群 ${batch.groupId} 会话画像 [${batch.startTime}, ${batch.endTime}) " +
-                                "被模型拒绝，已停止重试",
-                        cause,
-                    )
+                    if (requestController == null) {
+                        onRetryFailure(
+                            "群 ${batch.groupId} 会话画像 [${batch.startTime}, ${batch.endTime}) " +
+                                    "被模型拒绝，已停止重试",
+                            cause,
+                        )
+                    }
                    throw cause
                }
                lastFailure = cause
@@ -609,7 +662,7 @@ object UserProfileAnalysisService {
                            "第 ${attempt + 1}/$attempts 次分析失败",
                    cause = cause,
                    retryBackoff = retryBackoff,
-                    logFailure = onRetryFailure,
+                    logFailure = if (requestController == null) onRetryFailure else { _, _ -> },
                )
            }
        }
@@ -0,0 +1,258 @@
+package top.jie65535.mirai.profile
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancelAndJoin
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.withTimeoutOrNull
+import java.io.IOException
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class ProfileDailyRequestControllerTest {
+    @Test
+    fun growsHealthyAdmissionByFourWithoutExceedingTheConfiguredCap() = runBlocking {
+        val controller = controller(maxConcurrency = 128)
+
+        succeed(controller, key(1))
+        assertEquals(4, controller.snapshot().finalLimit)
+
+        repeat(4) { index -> succeed(controller, key(index + 2)) }
+        assertEquals(16, controller.snapshot().finalLimit)
+
+        repeat(16) { index -> succeed(controller, key(index + 6)) }
+        assertEquals(64, controller.snapshot().finalLimit)
+
+        repeat(64) { index -> succeed(controller, key(index + 22)) }
+        val stats = controller.snapshot()
+        assertEquals(128, stats.finalLimit)
+        assertEquals(128, stats.peakLimit)
+        assertFalse(stats.stopped)
+    }
+
+    @Test
+    fun pausesFreshAdmissionUntilTheFailedTaskRecovers() = runBlocking {
+        val controller = controller(maxConcurrency = 4)
+        succeed(controller, key(1))
+        val failedKey = key(2)
+        fail(controller, failedKey, attempt = 1)
+        val freshStarted = CompletableDeferred<Unit>()
+
+        coroutineScope {
+            val fresh = async {
+                controller.execute(key(3), attempt = 1) {
+                    freshStarted.complete(Unit)
+                }
+            }
+            assertNull(withTimeoutOrNull(50) { freshStarted.await() })
+
+            succeed(controller, failedKey, attempt = 2)
+            withTimeout(1_000) { freshStarted.await() }
+            fresh.await()
+        }
+
+        val stats = controller.snapshot()
+        assertEquals(1, stats.countedFailures)
+        assertEquals(1, stats.recoveredIncidents)
+        assertEquals(1, stats.maxParallelProbes)
+        assertFalse(stats.stopped)
+    }
+
+    @Test
+    fun admitsNewWorkWhenOneFastRequestFinishesBeforeSlowRequests() = runBlocking {
+        val controller = controller(maxConcurrency = 4)
+        succeed(controller, key(1))
+        val allEntered = CompletableDeferred<Unit>()
+        val releaseFast = CompletableDeferred<Unit>()
+        val releaseSlow = CompletableDeferred<Unit>()
+        val entered = AtomicInteger()
+
+        coroutineScope {
+            val activeRequests = (2..5).map { index ->
+                async {
+                    controller.execute(key(index), attempt = 1) {
+                        if (entered.incrementAndGet() == 4) allEntered.complete(Unit)
+                        if (index == 2) releaseFast.await() else releaseSlow.await()
+                    }
+                }
+            }
+            withTimeout(1_000) { allEntered.await() }
+
+            val newWorkStarted = CompletableDeferred<Unit>()
+            val queued = async {
+                controller.execute(key(6), attempt = 1) {
+                    newWorkStarted.complete(Unit)
+                }
+            }
+            assertNull(withTimeoutOrNull(50) { newWorkStarted.await() })
+            releaseFast.complete(Unit)
+            withTimeout(1_000) { newWorkStarted.await() }
+            releaseSlow.complete(Unit)
+            activeRequests.forEach { it.await() }
+            queued.await()
+        }
+
+        assertFalse(controller.snapshot().stopped)
+    }
+
+    @Test
+    fun keepsMultipleFailedTaskRetriesSingleFlight() = runBlocking {
+        val controller = controller(maxConcurrency = 4)
+        succeed(controller, key(1))
+        val failedKeys = listOf(key(2), key(3))
+        val allInitialAttemptsEntered = CompletableDeferred<Unit>()
+        val releaseInitialAttempts = CompletableDeferred<Unit>()
+        val initialEntered = AtomicInteger()
+
+        coroutineScope {
+            val initialAttempts = (failedKeys + key(4) + key(5)).mapIndexed { index, requestKey ->
+                async {
+                    runCatching {
+                        controller.execute(requestKey, attempt = 1) {
+                            if (initialEntered.incrementAndGet() == 4) {
+                                allInitialAttemptsEntered.complete(Unit)
+                            }
+                            releaseInitialAttempts.await()
+                            if (index < 2) throw IOException("transient")
+                        }
+                    }
+                }
+            }
+            withTimeout(1_000) { allInitialAttemptsEntered.await() }
+            releaseInitialAttempts.complete(Unit)
+            initialAttempts.forEach { it.await() }
+
+            val activeRetries = AtomicInteger()
+            val peakRetries = AtomicInteger()
+            val firstRetryEntered = CompletableDeferred<Unit>()
+            val releaseRetries = CompletableDeferred<Unit>()
+            val retries = failedKeys.map { requestKey ->
+                async {
+                    controller.execute(requestKey, attempt = 2) {
+                        val current = activeRetries.incrementAndGet()
+                        peakRetries.updateAndGet { previous -> maxOf(previous, current) }
+                        firstRetryEntered.complete(Unit)
+                        releaseRetries.await()
+                        activeRetries.decrementAndGet()
+                    }
+                }
+            }
+            withTimeout(1_000) { firstRetryEntered.await() }
+            delay(50)
+            assertEquals(1, peakRetries.get())
+            releaseRetries.complete(Unit)
+            retries.forEach { it.await() }
+        }
+
+        val stats = controller.snapshot()
+        assertEquals(1, stats.maxParallelProbes)
+        assertEquals(0, stats.exhaustedTasks)
+        assertFalse(stats.stopped)
+    }
+
+    @Test
+    fun stopsAfterThreeDifferentTasksExhaustTheirAttempts() = runBlocking {
+        val controller = controller(maxConcurrency = 128)
+
+        repeat(3) { taskIndex ->
+            repeat(3) { attemptIndex ->
+                fail(controller, key(taskIndex + 1), attemptIndex + 1)
+            }
+        }
+
+        val stats = controller.snapshot()
+        assertTrue(stats.stopped)
+        assertEquals(9, stats.totalAttempts)
+        assertEquals(9, stats.countedFailures)
+        assertEquals(8, stats.probeAttempts)
+        assertEquals(3, stats.exhaustedTasks)
+        assertFailsWith<ProfileDailyRunStoppedException> {
+            controller.execute(key(4), attempt = 1) { }
+        }
+        Unit
+    }
+
+    @Test
+    fun onePersistentBadTaskDoesNotStopHealthyWork() = runBlocking {
+        val controller = controller(maxConcurrency = 128)
+        repeat(3) { attemptIndex -> fail(controller, key(1), attemptIndex + 1) }
+
+        assertFalse(controller.snapshot().stopped)
+        succeed(controller, key(2))
+        succeed(controller, key(3))
+        succeed(controller, key(4))
+
+        val stats = controller.snapshot()
+        assertFalse(stats.stopped)
+        assertEquals(1, stats.exhaustedTasks)
+        assertEquals(1, stats.recoveredIncidents)
+    }
+
+    @Test
+    fun cancellationWhileWaitingDoesNotLeaveAQueuedRequest() = runBlocking {
+        val controller = controller(maxConcurrency = 1)
+        val firstEntered = CompletableDeferred<Unit>()
+        val releaseFirst = CompletableDeferred<Unit>()
+
+        coroutineScope {
+            val first = async {
+                controller.execute(key(1), attempt = 1) {
+                    firstEntered.complete(Unit)
+                    releaseFirst.await()
+                }
+            }
+            firstEntered.await()
+            val cancelled = async {
+                controller.execute(key(2), attempt = 1) {
+                    error("cancelled waiter must not start")
+                }
+            }
+            delay(50)
+            cancelled.cancelAndJoin()
+            releaseFirst.complete(Unit)
+            first.await()
+        }
+
+        succeed(controller, key(3))
+        assertEquals(2, controller.snapshot().totalAttempts)
+    }
+
+    private fun controller(maxConcurrency: Int) = ProfileDailyRequestController(
+        maxConcurrentRequests = maxConcurrency,
+        maxAttempts = 3,
+        growthFactor = 4,
+        maxConsecutiveExhaustedTasks = 3,
+    )
+
+    private suspend fun succeed(
+        controller: ProfileDailyRequestController,
+        key: ProfileDailyRequestKey,
+        attempt: Int = 1,
+    ) {
+        controller.execute(key, attempt) { }
+    }
+
+    private suspend fun fail(
+        controller: ProfileDailyRequestController,
+        key: ProfileDailyRequestKey,
+        attempt: Int,
+    ) {
+        assertFailsWith<IOException> {
+            controller.execute(key, attempt) { throw IOException("unavailable") }
+        }
+    }
+
+    private fun key(index: Int) = ProfileDailyRequestKey(
+        groupId = index.toLong(),
+        startTime = index * 100,
+        endTime = index * 100 + 50,
+    )
+}
@@ -142,6 +142,31 @@ class UserProfileAnalysisServiceTest {
        assertFalse(UserProfileStore.isConversationProcessed(INPUT_HASH))
    }

+    @Test
+    fun dailyControllerDoesNotCountSafetyRejectionAsAnOutage() = withProfileStore {
+        val controller = ProfileDailyRequestController(
+            maxConcurrentRequests = 128,
+            maxAttempts = 3,
+        )
+        val model = FakeConversationProfileModel {
+            throw ModelSafetyRejectionException(
+                errorType = "invalid_request_error",
+                errorCode = "cyber_policy",
+                message = "blocked by policy",
+            )
+        }
+
+        assertFailsWith<ModelSafetyRejectionException> {
+            analyze(batch(), model, retryMax = 2, requestController = controller)
+        }
+
+        val stats = controller.snapshot()
+        assertEquals(1, stats.totalAttempts)
+        assertEquals(0, stats.countedFailures)
+        assertEquals(0, stats.pauseIncidents)
+        assertFalse(stats.stopped)
+    }
+
    @Test
    fun skipsConversationAlreadyProcessedByAnEarlierCall() = withProfileStore {
        val model = FakeConversationProfileModel { successfulResult() }
@@ -250,12 +275,14 @@ class UserProfileAnalysisServiceTest {
        model: ConversationProfileModel,
        minAuthoredTextChars: Int = 1,
        retryMax: Int = 0,
+        requestController: ProfileDailyRequestController? = null,
    ): ConversationProfileAnalysisReport? = UserProfileAnalysisService.analyzeConversationBatch(
        batch = batch,
        minAuthoredTextChars = minAuthoredTextChars,
        model = model,
        retryMax = retryMax,
        summaryMaxLength = 500,
+        requestController = requestController,
    )

    private fun successfulResult() = result(