test(android): establish workflow runner

2026-07-25 18:47:22 +08:00
parent d013ac8c84
commit d57c4ae2d4
8 changed files with 548 additions and 7 deletions
@@ -87,6 +87,7 @@ dependencies {

    // Unit tests
    testImplementation("junit:junit:4.13.2")
+    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.7.3")

    // Debug
    debugImplementation("androidx.compose.ui:ui-tooling")
@@ -0,0 +1,79 @@
+package com.roubao.autopilot.workflow
+
+const val MAX_RETRIES_PER_STEP = 3
+
+data class WorkflowStep(
+    val id: String,
+    val timeoutMillis: Long,
+    val maxRetries: Int = 0
+) {
+    init {
+        require(id.isNotBlank()) { "Workflow step id must not be blank" }
+        require(timeoutMillis > 0) { "Workflow step timeout must be positive" }
+        require(maxRetries in 0..MAX_RETRIES_PER_STEP) {
+            "Workflow step maxRetries must be between 0 and $MAX_RETRIES_PER_STEP"
+        }
+    }
+}
+
+enum class WorkflowState {
+    IDLE,
+    RUNNING,
+    RETRYING,
+    SUCCEEDED,
+    FAILED,
+    BLOCKED,
+    STOPPED
+}
+
+enum class WorkflowFailureCode {
+    TIMEOUT,
+    TRANSIENT_AUTOMATION,
+    AUTOMATION_EXCEPTION
+}
+
+enum class SafetyStopReason {
+    LOGIN_REQUIRED,
+    VERIFICATION_REQUIRED,
+    RISK_CONTROL,
+    UNKNOWN_PAGE,
+    PAYMENT_BOUNDARY
+}
+
+sealed interface AutomationResult {
+    data object Success : AutomationResult
+
+    data class RetryableFailure(
+        val code: WorkflowFailureCode
+    ) : AutomationResult
+
+    data class FatalFailure(
+        val code: WorkflowFailureCode
+    ) : AutomationResult
+
+    data class Blocked(
+        val reason: SafetyStopReason
+    ) : AutomationResult
+
+    data object Stopped : AutomationResult
+}
+
+fun interface AutomationGateway {
+    suspend fun execute(step: WorkflowStep): AutomationResult
+}
+
+data class WorkflowTransition(
+    val from: WorkflowState,
+    val to: WorkflowState,
+    val stepId: String?,
+    val attempt: Int?
+)
+
+data class WorkflowReport(
+    val state: WorkflowState,
+    val completedStepIds: List<String>,
+    val attemptsByStep: Map<String, Int>,
+    val transitions: List<WorkflowTransition>,
+    val failureCode: WorkflowFailureCode? = null,
+    val safetyStopReason: SafetyStopReason? = null
+)
@@ -0,0 +1,207 @@
+package com.roubao.autopilot.workflow
+
+import java.util.concurrent.atomic.AtomicBoolean
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.TimeoutCancellationException
+import kotlinx.coroutines.async
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.supervisorScope
+import kotlinx.coroutines.withTimeout
+
+class WorkflowRunner(
+    private val automation: AutomationGateway
+) {
+    private val running = AtomicBoolean(false)
+    private val mutableState = MutableStateFlow(WorkflowState.IDLE)
+
+    @Volatile
+    private var stopRequested = false
+
+    @Volatile
+    private var currentExecution: Job? = null
+
+    val state: StateFlow<WorkflowState> = mutableState.asStateFlow()
+
+    suspend fun run(steps: List<WorkflowStep>): WorkflowReport {
+        require(steps.isNotEmpty()) { "Workflow must contain at least one step" }
+        check(running.compareAndSet(false, true)) { "WorkflowRunner is already running" }
+
+        stopRequested = false
+        val completedSteps = mutableListOf<String>()
+        val attempts = linkedMapOf<String, Int>()
+        val transitions = mutableListOf<WorkflowTransition>()
+
+        if (mutableState.value != WorkflowState.IDLE) {
+            transition(WorkflowState.IDLE, null, null, transitions)
+        }
+
+        try {
+            for (step in steps) {
+                var attempt = 0
+                while (attempt <= step.maxRetries) {
+                    if (stopRequested) {
+                        return terminalReport(
+                            state = WorkflowState.STOPPED,
+                            completedSteps = completedSteps,
+                            attempts = attempts,
+                            transitions = transitions,
+                            step = step,
+                            attempt = attempt.takeIf { it > 0 }
+                        )
+                    }
+
+                    attempt += 1
+                    attempts[step.id] = attempt
+                    transition(WorkflowState.RUNNING, step.id, attempt, transitions)
+
+                    when (val result = executeStep(step)) {
+                        AutomationResult.Success -> {
+                            completedSteps += step.id
+                            break
+                        }
+
+                        is AutomationResult.Blocked -> {
+                            return terminalReport(
+                                state = WorkflowState.BLOCKED,
+                                completedSteps = completedSteps,
+                                attempts = attempts,
+                                transitions = transitions,
+                                step = step,
+                                attempt = attempt,
+                                safetyStopReason = result.reason
+                            )
+                        }
+
+                        is AutomationResult.FatalFailure -> {
+                            return terminalReport(
+                                state = WorkflowState.FAILED,
+                                completedSteps = completedSteps,
+                                attempts = attempts,
+                                transitions = transitions,
+                                step = step,
+                                attempt = attempt,
+                                failureCode = result.code
+                            )
+                        }
+
+                        is AutomationResult.RetryableFailure -> {
+                            if (attempt > step.maxRetries) {
+                                return terminalReport(
+                                    state = WorkflowState.FAILED,
+                                    completedSteps = completedSteps,
+                                    attempts = attempts,
+                                    transitions = transitions,
+                                    step = step,
+                                    attempt = attempt,
+                                    failureCode = result.code
+                                )
+                            }
+                            transition(WorkflowState.RETRYING, step.id, attempt, transitions)
+                        }
+
+                        AutomationResult.Stopped -> {
+                            return terminalReport(
+                                state = WorkflowState.STOPPED,
+                                completedSteps = completedSteps,
+                                attempts = attempts,
+                                transitions = transitions,
+                                step = step,
+                                attempt = attempt
+                            )
+                        }
+                    }
+                }
+            }
+
+            return terminalReport(
+                state = WorkflowState.SUCCEEDED,
+                completedSteps = completedSteps,
+                attempts = attempts,
+                transitions = transitions,
+                step = null,
+                attempt = null
+            )
+        } finally {
+            currentExecution = null
+            running.set(false)
+        }
+    }
+
+    fun requestStop() {
+        stopRequested = true
+        currentExecution?.cancel(CancellationException("Workflow stop requested"))
+    }
+
+    private suspend fun executeStep(step: WorkflowStep): AutomationResult = supervisorScope {
+        val execution: Deferred<AutomationResult> = async {
+            try {
+                automation.execute(step)
+            } catch (error: CancellationException) {
+                throw error
+            } catch (_: Exception) {
+                AutomationResult.FatalFailure(WorkflowFailureCode.AUTOMATION_EXCEPTION)
+            }
+        }
+        currentExecution = execution
+
+        try {
+            withTimeout(step.timeoutMillis) {
+                execution.await()
+            }
+        } catch (_: TimeoutCancellationException) {
+            AutomationResult.RetryableFailure(WorkflowFailureCode.TIMEOUT)
+        } catch (error: CancellationException) {
+            if (stopRequested) {
+                AutomationResult.Stopped
+            } else {
+                throw error
+            }
+        } finally {
+            execution.cancel()
+            if (currentExecution === execution) {
+                currentExecution = null
+            }
+        }
+    }
+
+    private fun terminalReport(
+        state: WorkflowState,
+        completedSteps: List<String>,
+        attempts: Map<String, Int>,
+        transitions: MutableList<WorkflowTransition>,
+        step: WorkflowStep?,
+        attempt: Int?,
+        failureCode: WorkflowFailureCode? = null,
+        safetyStopReason: SafetyStopReason? = null
+    ): WorkflowReport {
+        transition(state, step?.id, attempt, transitions)
+        return WorkflowReport(
+            state = state,
+            completedStepIds = completedSteps.toList(),
+            attemptsByStep = attempts.toMap(),
+            transitions = transitions.toList(),
+            failureCode = failureCode,
+            safetyStopReason = safetyStopReason
+        )
+    }
+
+    private fun transition(
+        next: WorkflowState,
+        stepId: String?,
+        attempt: Int?,
+        transitions: MutableList<WorkflowTransition>
+    ) {
+        val previous = mutableState.value
+        mutableState.value = next
+        transitions += WorkflowTransition(
+            from = previous,
+            to = next,
+            stepId = stepId,
+            attempt = attempt
+        )
+    }
+}
@@ -0,0 +1,162 @@
+package com.roubao.autopilot.workflow
+
+import java.util.ArrayDeque
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.test.currentTime
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Assert.fail
+import org.junit.Test
+
+class WorkflowRunnerTest {
+    @Test
+    fun `multiple steps complete with explicit transitions`() = runTest {
+        val automation = QueueAutomation(
+            AutomationResult.Success,
+            AutomationResult.Success
+        )
+        val runner = WorkflowRunner(automation)
+
+        val report = runner.run(
+            listOf(
+                step("open_app"),
+                step("verify_page")
+            )
+        )
+
+        assertEquals(WorkflowState.SUCCEEDED, report.state)
+        assertEquals(listOf("open_app", "verify_page"), report.completedStepIds)
+        assertEquals(mapOf("open_app" to 1, "verify_page" to 1), report.attemptsByStep)
+        assertEquals(
+            listOf(
+                WorkflowState.RUNNING,
+                WorkflowState.RUNNING,
+                WorkflowState.SUCCEEDED
+            ),
+            report.transitions.map { it.to }
+        )
+    }
+
+    @Test
+    fun `retryable failure retries once then succeeds`() = runTest {
+        val automation = QueueAutomation(
+            AutomationResult.RetryableFailure(WorkflowFailureCode.TRANSIENT_AUTOMATION),
+            AutomationResult.Success
+        )
+        val runner = WorkflowRunner(automation)
+
+        val report = runner.run(listOf(step("search", maxRetries = 1)))
+
+        assertEquals(WorkflowState.SUCCEEDED, report.state)
+        assertEquals(2, report.attemptsByStep["search"])
+        assertEquals(
+            listOf(
+                WorkflowState.RUNNING,
+                WorkflowState.RETRYING,
+                WorkflowState.RUNNING,
+                WorkflowState.SUCCEEDED
+            ),
+            report.transitions.map { it.to }
+        )
+    }
+
+    @Test
+    fun `timeout is bounded and fails after retry budget`() = runTest {
+        val runner = WorkflowRunner {
+            delay(10_000)
+            AutomationResult.Success
+        }
+
+        val report = runner.run(
+            listOf(step("wait_for_page", timeoutMillis = 100, maxRetries = 1))
+        )
+
+        assertEquals(WorkflowState.FAILED, report.state)
+        assertEquals(WorkflowFailureCode.TIMEOUT, report.failureCode)
+        assertEquals(2, report.attemptsByStep["wait_for_page"])
+        assertEquals(200, currentTime)
+    }
+
+    @Test
+    fun `all safety reasons block without retry`() = runTest {
+        SafetyStopReason.entries.forEach { reason ->
+            var calls = 0
+            val runner = WorkflowRunner {
+                calls += 1
+                AutomationResult.Blocked(reason)
+            }
+
+            val report = runner.run(listOf(step("guard", maxRetries = 3)))
+
+            assertEquals(WorkflowState.BLOCKED, report.state)
+            assertEquals(reason, report.safetyStopReason)
+            assertEquals(1, calls)
+        }
+    }
+
+    @Test
+    fun `user stop cancels current child and returns stopped report`() = runTest {
+        val started = CompletableDeferred<Unit>()
+        val runner = WorkflowRunner {
+            started.complete(Unit)
+            awaitCancellation()
+        }
+
+        val runningReport = async {
+            runner.run(listOf(step("long_action", timeoutMillis = 60_000)))
+        }
+        started.await()
+        runner.requestStop()
+
+        val report = runningReport.await()
+        assertEquals(WorkflowState.STOPPED, report.state)
+        assertTrue(coroutineContext.isActive)
+    }
+
+    @Test
+    fun `concurrent run is rejected`() = runTest {
+        val started = CompletableDeferred<Unit>()
+        val runner = WorkflowRunner {
+            started.complete(Unit)
+            awaitCancellation()
+        }
+        val firstRun = async {
+            runner.run(listOf(step("first", timeoutMillis = 60_000)))
+        }
+        started.await()
+
+        try {
+            runner.run(listOf(step("second")))
+            fail("Expected concurrent run to be rejected")
+        } catch (error: IllegalStateException) {
+            assertEquals("WorkflowRunner is already running", error.message)
+        } finally {
+            runner.requestStop()
+            assertEquals(WorkflowState.STOPPED, firstRun.await().state)
+        }
+    }
+
+    private fun step(
+        id: String,
+        timeoutMillis: Long = 1_000,
+        maxRetries: Int = 0
+    ) = WorkflowStep(
+        id = id,
+        timeoutMillis = timeoutMillis,
+        maxRetries = maxRetries
+    )
+
+    private class QueueAutomation(
+        vararg results: AutomationResult
+    ) : AutomationGateway {
+        private val remaining = ArrayDeque(results.toList())
+
+        override suspend fun execute(step: WorkflowStep): AutomationResult =
+            remaining.removeFirst()
+    }
+}