feat(scene): complete automation runtime #11

Merged
KremeCN merged 9 commits from feat/scene-automation-runtime into main 2026-08-28 06:39:31 +08:00
3 changed files with 38 additions and 22 deletions
Showing only changes of commit c4af3266be - Show all commits
@@ -38,6 +38,7 @@ data class ScheduledAutomationOccurrence(
val sceneId: String,
val triggerId: String,
val occurrenceKey: String,
val triggerDefinition: String,
val scheduledAtEpochMillis: Long,
val expiresAtEpochMillis: Long,
)
@@ -57,23 +58,37 @@ class DefaultAutomationEngine(
override suspend fun evaluate(
profileId: ProviderProfileId,
scheduledOccurrence: ScheduledAutomationOccurrence?,
): DomainResult<Int> = evaluationMutex.withLock {
Review

执行阻塞

evaluationMutex 覆盖了整个评估过程,其中还会等待场景动作执行完成。包含 Delay 或慢速设备 I/O 的场景会长期占用这个全局锁,阻塞前台轮询和其他 WorkManager 评估,可能造成触发窗口错过。应在完成快照评估和原子占用后释放锁,再执行动作。

**执行阻塞** `evaluationMutex` 覆盖了整个评估过程,其中还会等待场景动作执行完成。包含 `Delay` 或慢速设备 I/O 的场景会长期占用这个全局锁,阻塞前台轮询和其他 WorkManager 评估,可能造成触发窗口错过。应在完成快照评估和原子占用后释放锁,再执行动作。 <!-- pr-agent-dedup: 75e5331a6c95 --> <!-- pr-agent-key-issue-location: 42b5e002dc41 -->
evaluateLocked(profileId, scheduledOccurrence)
): DomainResult<Int> {
val scenesToExecute = when (val prepared = evaluationMutex.withLock { prepareExecutions(profileId, scheduledOccurrence) }) {
is DomainResult.Failure -> return prepared
is DomainResult.Success -> prepared.value
}
var executed = 0
for (scene in scenesToExecute) {
sceneExecutor.execute(
scene = scene,
executionId = SceneExecutionId("auto-${UUID.randomUUID()}"),
source = SceneExecutionSource.AUTOMATION,
).collect { update ->
if (update is com.flagship.abox.manager.domain.SceneExecutionUpdate.Completed) executed++
}
}
return DomainResult.Success(executed)
}
private suspend fun evaluateLocked(
private suspend fun prepareExecutions(
profileId: ProviderProfileId,
scheduledOccurrence: ScheduledAutomationOccurrence?,
): DomainResult<Int> {
): DomainResult<List<Scene>> {
val scenes = when (val result = sceneRepository.observeScenes(profileId).first()) {
is DomainResult.Failure -> return result
is DomainResult.Success -> result.value.filter { it.enabled && it.type == AutomationType.AUTOMATIC }
}
if (scenes.isEmpty()) return DomainResult.Success(0)
if (scenes.isEmpty()) return DomainResult.Success(emptyList())
val snapshot = collectSnapshot(profileId, scenes)
val now = LocalDateTime.ofInstant(Instant.ofEpochMilli(snapshot.observedAtEpochMillis), ZoneId.systemDefault())
var executed = 0
val executions = mutableListOf<Scene>()
for (scene in scenes) {
var newlyMatched = false
val newlyMatchedStateTriggers = mutableListOf<SceneTrigger>()
@@ -91,6 +106,7 @@ class DefaultAutomationEngine(
trigger is SceneTrigger.Time &&
scheduledOccurrence?.sceneId == scene.id.value &&
scheduledOccurrence.triggerId == trigger.id.value &&
scheduledOccurrence.triggerDefinition == trigger.automationDefinition() &&
snapshot.observedAtEpochMillis in scheduledOccurrence.scheduledAtEpochMillis..scheduledOccurrence.expiresAtEpochMillis
) {
Outdated
Review

边沿丢失

阈值刚越界但条件不满足时,这里会把刚写入的匹配状态重置为 false。如果阈值之后一直保持越界,只要条件在后续轮询中变为满足,场景就会在没有新阈值边沿的情况下执行,破坏边沿触发语义。条件失败后应保留已匹配状态,或显式建模待处理事件。

**边沿丢失** 阈值刚越界但条件不满足时,这里会把刚写入的匹配状态重置为 `false`。如果阈值之后一直保持越界,只要条件在后续轮询中变为满足,场景就会在没有新阈值边沿的情况下执行,破坏边沿触发语义。条件失败后应保留已匹配状态,或显式建模待处理事件。 <!-- pr-agent-dedup: a7138467d563 --> <!-- pr-agent-key-issue-location: 5ff8a986e04d -->
TriggerEvaluation.Matched(scheduledOccurrence.occurrenceKey)
@@ -143,16 +159,10 @@ class DefaultAutomationEngine(
if (executionCount == 0) continue
repeat(executionCount) {
sceneExecutor.execute(
scene = scene,
executionId = SceneExecutionId("auto-${UUID.randomUUID()}"),
source = SceneExecutionSource.AUTOMATION,
).collect { update ->
Review

重复执行

当状态触发器与时间触发器在同一轮同时命中,或多个时间触发器同时命中时,executionCount 会大于 1,导致同一场景的全部动作被重复执行。插座、红外等非幂等动作可能因此连续执行多次;同一轮评估中只应将场景加入一次,除非明确要求按触发器分别执行。

**重复执行** 当状态触发器与时间触发器在同一轮同时命中,或多个时间触发器同时命中时,`executionCount` 会大于 1,导致同一场景的全部动作被重复执行。插座、红外等非幂等动作可能因此连续执行多次;同一轮评估中只应将场景加入一次,除非明确要求按触发器分别执行。 <!-- pr-agent-dedup: c9d00b09d401 --> <!-- pr-agent-key-issue-location: e0ca46a9d596 -->
if (update is com.flagship.abox.manager.domain.SceneExecutionUpdate.Completed) executed++
}
executions += scene
}
}
return DomainResult.Success(executed)
return DomainResult.Success(executions)
}
private suspend fun collectSnapshot(profileId: ProviderProfileId, scenes: List<Scene>): AutomationSnapshot {
@@ -184,3 +194,6 @@ class DefaultAutomationEngine(
}
}
internal fun SceneTrigger.Time.automationDefinition(): String =
listOf(timeOfDay, daysOfWeek.sorted().joinToString(",")).joinToString("|")
@@ -37,10 +37,11 @@ class AutomationPollWorker @AssistedInject constructor(
val scheduledOccurrence = inputData.getString(WorkManagerAutomationScheduleStore.KEY_OCCURRENCE_KEY)?.let { key ->
val sceneId = inputData.getString(WorkManagerAutomationScheduleStore.KEY_SCENE_ID)
val triggerId = inputData.getString(WorkManagerAutomationScheduleStore.KEY_TRIGGER_ID)
val triggerDefinition = inputData.getString(WorkManagerAutomationScheduleStore.KEY_TRIGGER_DEFINITION)
val scheduledAt = inputData.getLong(WorkManagerAutomationScheduleStore.KEY_SCHEDULED_AT_EPOCH_MILLIS, -1L)
val expiresAt = inputData.getLong(WorkManagerAutomationScheduleStore.KEY_EXPIRES_AT_EPOCH_MILLIS, -1L)
if (sceneId != null && triggerId != null && scheduledAt >= 0L && expiresAt >= scheduledAt) {
ScheduledAutomationOccurrence(sceneId, triggerId, key, scheduledAt, expiresAt)
if (sceneId != null && triggerId != null && triggerDefinition != null && scheduledAt >= 0L && expiresAt >= scheduledAt) {
ScheduledAutomationOccurrence(sceneId, triggerId, key, triggerDefinition, scheduledAt, expiresAt)
} else null
}
when (automationEngine.evaluate(profileId, scheduledOccurrence)) {
@@ -24,7 +24,6 @@ class WorkManagerAutomationScheduleStore(context: Context) : AutomationScheduleS
.setInputData(androidx.work.Data.Builder().putString(KEY_PROFILE_ID, profileId.value).build())
.build()
workManager.enqueueUniquePeriodicWork(workName(profileId), ExistingPeriodicWorkPolicy.UPDATE, request)
workManager.cancelAllWorkByTag(timeWorkTag(profileId))
Review

任务中断

replace 每次协调都会按标签取消全部定时任务,这也会停止已经处于运行状态的 worker。若用户在定时场景执行期间保存、启用或删除任意场景,当前执行可能在触发记录已被占用后遭到取消,导致动作只执行一部分且无法再次触发。应仅替换尚未运行的调度,或避免取消正在运行的工作。

**任务中断** `replace` 每次协调都会按标签取消全部定时任务,这也会停止已经处于运行状态的 worker。若用户在定时场景执行期间保存、启用或删除任意场景,当前执行可能在触发记录已被占用后遭到取消,导致动作只执行一部分且无法再次触发。应仅替换尚未运行的调度,或避免取消正在运行的工作。 <!-- pr-agent-dedup: ecf27d5946b4 --> <!-- pr-agent-key-issue-location: 8c2cba4abb4d -->
Review

任务残留

replace 只添加当前时间触发器的任务;只要该配置仍有任一自动场景,就不会取消已删除、禁用或改时的旧任务。反复编辑会累积旧 worker,它们到点后仍会校验会话并评估整个配置,产生不必要的后台 I/O 和设备请求。替换计划时应清理或比对旧时间任务。

**任务残留** `replace` 只添加当前时间触发器的任务;只要该配置仍有任一自动场景,就不会取消已删除、禁用或改时的旧任务。反复编辑会累积旧 worker,它们到点后仍会校验会话并评估整个配置,产生不必要的后台 I/O 和设备请求。替换计划时应清理或比对旧时间任务。 <!-- pr-agent-dedup: 20f0c54dcf7a --> <!-- pr-agent-key-issue-location: eb9820b6b300 -->
Review

任务泄漏

replace 只为当前时间触发器添加任务,却不取消已删除、禁用或修改的触发器对应的旧一次性任务。只要仍有其他自动场景,旧任务就会一直保留到执行;反复编辑时间触发器会累积后台任务,并在触发时产生多余的会话验证、设备读取和场景评估。

**任务泄漏** `replace` 只为当前时间触发器添加任务,却不取消已删除、禁用或修改的触发器对应的旧一次性任务。只要仍有其他自动场景,旧任务就会一直保留到执行;反复编辑时间触发器会累积后台任务,并在触发时产生多余的会话验证、设备读取和场景评估。 <!-- pr-agent-dedup: 1812827f9fe7 --> <!-- pr-agent-key-issue-location: 8f72625368dd -->
scenes.forEach { scene ->
scene.triggers.filterIsInstance<SceneTrigger.Time>().forEach { trigger ->
scheduleNextTimeTrigger(profileId, scene, trigger)
@@ -40,8 +39,8 @@ class WorkManagerAutomationScheduleStore(context: Context) : AutomationScheduleS
}
private fun workName(profileId: ProviderProfileId) = "automation:${profileId.value}"
Outdated
Review

计划遗漏

整个配置档只保存最早的一个时间 occurrence。若 Doze 或电量限制使任务延迟超过二十分钟,期间其他已到期的时间触发器既没有独立的计划任务,也已超出普通评估的宽限期;重新协调时又只会查找未来时间,因此这些执行会被永久跳过。应为每个 occurrence 建立唯一任务,或持久化并依次处理全部待执行 occurrence。

**计划遗漏** 整个配置档只保存最早的一个时间 occurrence。若 Doze 或电量限制使任务延迟超过二十分钟,期间其他已到期的时间触发器既没有独立的计划任务,也已超出普通评估的宽限期;重新协调时又只会查找未来时间,因此这些执行会被永久跳过。应为每个 occurrence 建立唯一任务,或持久化并依次处理全部待执行 occurrence。 <!-- pr-agent-dedup: 44f30beb57cd --> <!-- pr-agent-key-issue-location: bdcee4cf7c5b -->
private fun timeWorkName(profileId: ProviderProfileId, scene: Scene, trigger: SceneTrigger.Time) =
"automation-time:${profileId.value}:${scene.id.value}:${trigger.id.value}"
private fun timeWorkName(profileId: ProviderProfileId, scene: Scene, trigger: SceneTrigger.Time, occurrenceKey: String) =
"automation-time:${profileId.value}:${scene.id.value}:${trigger.id.value}:$occurrenceKey"
private fun timeWorkTag(profileId: ProviderProfileId) = "automation-time:${profileId.value}"
private fun scheduleNextTimeTrigger(profileId: ProviderProfileId, scene: Scene, trigger: SceneTrigger.Time) {
@@ -55,6 +54,7 @@ class WorkManagerAutomationScheduleStore(context: Context) : AutomationScheduleS
?.toInstant()
?.minusMillis(1)
val expiresAt = listOfNotNull(maximumExpiry, followingOccurrence).minOrNull() ?: maximumExpiry
val occurrenceKey = "${scheduledAt.toLocalDate()}:${trigger.timeOfDay}:${trigger.id.value}"
val request = OneTimeWorkRequestBuilder<AutomationPollWorker>()
.setInitialDelay(delay.toMillis().coerceAtLeast(0), TimeUnit.MILLISECONDS)
.addTag(timeWorkTag(profileId))
@@ -63,15 +63,16 @@ class WorkManagerAutomationScheduleStore(context: Context) : AutomationScheduleS
.putString(KEY_PROFILE_ID, profileId.value)
.putString(KEY_SCENE_ID, scene.id.value)
.putString(KEY_TRIGGER_ID, trigger.id.value)
.putString(KEY_OCCURRENCE_KEY, "${scheduledAt.toLocalDate()}:${trigger.timeOfDay}:${trigger.id.value}")
.putString(KEY_OCCURRENCE_KEY, occurrenceKey)
.putString(KEY_TRIGGER_DEFINITION, trigger.automationDefinition())
.putLong(KEY_SCHEDULED_AT_EPOCH_MILLIS, scheduledAtInstant.toEpochMilli())
.putLong(KEY_EXPIRES_AT_EPOCH_MILLIS, expiresAt.toEpochMilli())
.build(),
)
.build()
workManager.enqueueUniqueWork(
timeWorkName(profileId, scene, trigger),
androidx.work.ExistingWorkPolicy.REPLACE,
timeWorkName(profileId, scene, trigger, occurrenceKey),
androidx.work.ExistingWorkPolicy.KEEP,
request,
)
}
@@ -92,6 +93,7 @@ class WorkManagerAutomationScheduleStore(context: Context) : AutomationScheduleS
const val KEY_SCENE_ID = "scene_id"
const val KEY_TRIGGER_ID = "trigger_id"
const val KEY_OCCURRENCE_KEY = "occurrence_key"
const val KEY_TRIGGER_DEFINITION = "trigger_definition"
const val KEY_SCHEDULED_AT_EPOCH_MILLIS = "scheduled_at_epoch_millis"
const val KEY_EXPIRES_AT_EPOCH_MILLIS = "expires_at_epoch_millis"