diff --git a/app/src/main/java/com/yugahashimoto/andcode/AndCodeApplication.kt b/app/src/main/java/com/yugahashimoto/andcode/AndCodeApplication.kt index 71c51e2dc..5c93b19d2 100644 --- a/app/src/main/java/com/yugahashimoto/andcode/AndCodeApplication.kt +++ b/app/src/main/java/com/yugahashimoto/andcode/AndCodeApplication.kt @@ -581,9 +581,15 @@ class AndCodeApplication : Application() { // Re-arm alarms for schedules that were saved in a previous process lifetime. scheduleManager.rescheduleAll() // Let guest agents read and manage schedules through the and-code-schedule MCP server. - // The poll loop is a cheap no-op (one directory stat) while no guest has touched the bridge. + // The guest can only be asking for something while it can run: the app is on screen, or + // something holds runtimeWork (and with it the wake lock). With neither, the device is free + // to suspend and the guest freezes with it, so the loop stays parked instead of polling for + // the life of the process. val scheduleBridge = ScheduleBridge(File(runtimeDirectory, "workspace"), AppScheduleStore(scheduleRepository, scheduleManager)) - applicationScope.launch { scheduleBridge.run() } + val scheduleBridgeActive = + combine(appForeground.foreground, runtimeWork.active) { foreground, working -> foreground || working } + .distinctUntilChanged() + applicationScope.launch { scheduleBridge.run(scheduleBridgeActive) } observeForegroundForRuntimeRestart() scheduleDeferredInitialization() } diff --git a/app/src/main/java/com/yugahashimoto/andcode/core/util/FlowExt.kt b/app/src/main/java/com/yugahashimoto/andcode/core/util/FlowExt.kt index 1a9e907c9..127a3602c 100644 --- a/app/src/main/java/com/yugahashimoto/andcode/core/util/FlowExt.kt +++ b/app/src/main/java/com/yugahashimoto/andcode/core/util/FlowExt.kt @@ -2,8 +2,10 @@ package com.yugahashimoto.andcode.core.util import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.channelFlow +import kotlinx.coroutines.flow.collectLatest import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeoutOrNull @@ -79,3 +81,34 @@ fun Flow.debounceFalseEdge(graceMillis: Long): Flow = if (values.isClosedForReceive) break } } + +/** + * Runs [action] straight away and then every [intervalMillis], for as long as the latest value of + * this flow is `true`; while it is `false` nothing runs and no timer is armed. + * + * Written for the file polls that sit between the app and the in-guest agent (the permission + * bridge, the schedule bridge, the browser command file). Each used to loop for the life of the + * process, so an app that had run once kept waking every few hundred milliseconds to stat a + * directory nobody could have written to - the guest only runs while the device is awake for it, and + * that is exactly when one of these gates is open. Parking the loop on a flow turns "poll forever" + * into "poll while something could be there". + * + * Opening the gate polls at once rather than after a first interval, so whatever arrived while it + * was shut - a request written just before the app was backgrounded, say - is picked up on the way + * back in instead of a beat later. Closing it cancels the wait in progress, so nothing fires after + * the flip. An [action] that never suspends always runs to completion; one that does is cancelled + * at its next suspension point, so it should not suspend between taking something and handing it + * on. Never returns normally while the flow stays open, so callers launch it in a scope they cancel. + */ +suspend fun Flow.pollWhileActive( + intervalMillis: Long, + action: suspend () -> Unit, +) { + collectLatest { active -> + if (!active) return@collectLatest + while (true) { + action() + delay(intervalMillis) + } + } +} diff --git a/app/src/main/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcher.kt b/app/src/main/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcher.kt index 60d13457d..06011572e 100644 --- a/app/src/main/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcher.kt +++ b/app/src/main/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcher.kt @@ -5,9 +5,10 @@ import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.rememberUpdatedState import com.yugahashimoto.andcode.core.storage.DeviceStorage +import com.yugahashimoto.andcode.core.util.pollWhileActive import com.yugahashimoto.andcode.runtime.LocalAgent import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.withContext import java.io.File @@ -17,26 +18,51 @@ private const val POLL_INTERVAL_MILLIS = 1000L * Watches the active workspace for a browser command written by the in-guest agent * (`.and-code/browser-command.json`, e.g. `{"action":"open","url":"http://127.0.0.1:8080/"}`) * and opens the guest browser at the requested URL so the user can watch and join in. + * + * Only polls while [foreground] is `true`: opening a browser nobody is looking at achieves nothing, + * and a composition outlives the activity being stopped, so without the gate this read the disk + * every second for as long as the process was alive. A command written while the app was away is + * opened as soon as it returns. + * + * [foreground] keys the effect, so it must be a reference that stays the same across recompositions + * (the app's own `StateFlow` is); a flow built inline would restart the watcher, and its immediate + * first poll, on every recomposition. */ @Composable fun GuestBrowserCommandWatcher( workspacePath: String?, agent: LocalAgent?, runtimeDirectory: File, + foreground: Flow, onOpenUrl: (String) -> Unit, ) { val openUrl by rememberUpdatedState(onOpenUrl) - LaunchedEffect(workspacePath, agent, runtimeDirectory) { + LaunchedEffect(workspacePath, agent, runtimeDirectory, foreground) { if (workspacePath == null || agent == null) return@LaunchedEffect - while (true) { - delay(POLL_INTERVAL_MILLIS) - val url = + watchBrowserCommands( + active = foreground, + poll = { withContext(Dispatchers.IO) { consumeBrowserOpenCommand(workspacePath, agent, runtimeDirectory, DeviceStorage.mounts()) } - if (url != null) { - openUrl(url) - } - } + }, + onOpenUrl = { url -> openUrl(url) }, + ) } } + +/** + * The watcher's loop, apart from Compose so the gating can be tested with a virtual clock. + * + * [poll] suspends while it reads the command file, so a gate that closes in that instant can drop + * the URL it has just consumed. That is a window of milliseconds at the moment the app is leaving + * the screen, when nobody is left to watch the browser open. + */ +internal suspend fun watchBrowserCommands( + active: Flow, + poll: suspend () -> String?, + onOpenUrl: (String) -> Unit, + intervalMillis: Long = POLL_INTERVAL_MILLIS, +) { + active.pollWhileActive(intervalMillis) { poll()?.let(onOpenUrl) } +} diff --git a/app/src/main/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridge.kt b/app/src/main/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridge.kt index f680cd302..83abed295 100644 --- a/app/src/main/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridge.kt +++ b/app/src/main/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridge.kt @@ -1,10 +1,11 @@ package com.yugahashimoto.andcode.feature.schedule +import com.yugahashimoto.andcode.core.util.pollWhileActive import com.yugahashimoto.andcode.data.schedule.CronExpression import com.yugahashimoto.andcode.data.schedule.Schedule import com.yugahashimoto.andcode.data.schedule.ScheduleRepository import com.yugahashimoto.andcode.data.schedule.ScheduleRun -import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow import org.json.JSONArray import org.json.JSONObject import java.io.File @@ -92,14 +93,17 @@ class ScheduleBridge( ) { val bridgeDir: File = File(workspaceDir, BRIDGE_RELATIVE_PATH) - /** Runs the poll loop until cancelled; typically launched on the application scope. */ - suspend fun run() { - while (true) { - // A single failing request or IO error must never kill the bridge for the process - // lifetime; drop it and keep polling. - runCatching { pollOnce() } - delay(pollIntervalMillis) - } + /** + * Runs the poll loop until cancelled; typically launched on the application scope. + * + * [active] says whether the guest could be talking to the bridge right now. It is required + * rather than defaulted to "always": a default would let a future call site forget it and + * quietly bring back a loop that polls for the life of the process. + */ + suspend fun run(active: Flow) { + // A single failing request or IO error must never kill the bridge for the process + // lifetime; drop it and keep polling. + active.pollWhileActive(pollIntervalMillis) { runCatching { pollOnce() } } } /** Handles every currently pending request and prunes stale files. */ @@ -129,6 +133,16 @@ class ScheduleBridge( writeResponse(responsesDir, requestId, errorResponse("Malformed request"), now) return } + // The guest stops waiting after RESPONSE_TIMEOUT_S and reports the call as failed. A request + // that sat unanswered past that - written while the poll was parked - must not run on the + // way back in, or a create/delete/runNow the agent was told did not happen would happen + // anyway. A missing or zero timestamp is "unknown", not "ancient", so it still runs. + val createdAtMs = request.optLong("createdAtMs", 0L) + if (createdAtMs > 0L && now - createdAtMs > REQUEST_EXPIRES_AFTER_MILLIS) { + file.delete() + writeResponse(responsesDir, requestId, errorResponse("Request expired before the AndCode app could handle it"), now) + return + } val response = try { execute(request) @@ -367,5 +381,13 @@ class ScheduleBridge( const val BRIDGE_RELATIVE_PATH = ".and-code/schedule-bridge" const val POLL_INTERVAL_MILLIS = 800L const val STALE_AFTER_MILLIS = 15 * 60_000L + + /** + * Just past the guest's 90 s reply timeout (`RESPONSE_TIMEOUT_S` in andcode-schedule-mcp.py, + * which must be kept in step with this), so a request is dropped once its caller has given up + * on it. No allowance for clock drift is needed: the guest runs under PRoot and reads the same + * kernel clock as this process. + */ + const val REQUEST_EXPIRES_AFTER_MILLIS = 95_000L } } diff --git a/app/src/main/java/com/yugahashimoto/andcode/runtime/local/ClaudeCodeRuntime.kt b/app/src/main/java/com/yugahashimoto/andcode/runtime/local/ClaudeCodeRuntime.kt index b61f3aa2c..d054d2aa2 100644 --- a/app/src/main/java/com/yugahashimoto/andcode/runtime/local/ClaudeCodeRuntime.kt +++ b/app/src/main/java/com/yugahashimoto/andcode/runtime/local/ClaudeCodeRuntime.kt @@ -8,13 +8,13 @@ import com.yugahashimoto.andcode.core.api.OpenCodeTime import com.yugahashimoto.andcode.core.api.OpenCodeTodo import com.yugahashimoto.andcode.core.api.PromptAttachment import com.yugahashimoto.andcode.core.api.QuestionRequest +import com.yugahashimoto.andcode.core.util.pollWhileActive import com.yugahashimoto.andcode.runtime.PermissionResponse import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow @@ -66,6 +66,14 @@ class ClaudeCodeRuntime( ClaudePermissionBridge(File(runtimeDirectory, ClaudePermissionBridge.HOST_DIR_NAME)) private var bridgeWatchJob: Job? = null + /** + * Whether any CLI process is registered in [sessions]; what the permission-bridge watcher polls + * on. Kept in step by [registerSession] and [unregisterSession], under the same monitor as + * [sessions], so the watcher can never be left parked while a process that could raise a request + * exists. + */ + private val hasLiveSessions = MutableStateFlow(false) + /** One live CLI process, plus what is needed to decide whether it can be reused. */ private class SessionProcess( val process: Process, @@ -78,6 +86,7 @@ class ClaudeCodeRuntime( val systemPrompt: String?, ) + /** Mutate only through [registerSession] and [unregisterSession], which keep [hasLiveSessions] in step. */ private val sessions = linkedMapOf() /** @@ -235,32 +244,67 @@ class ClaudeCodeRuntime( .filter { sessions[it.androidSessionId]?.process?.isAlive == true } .mapNotNull(permissionBridge::toQuestionRequest) + /** The only place [sessions] gains an entry, so [hasLiveSessions] can never lag behind it. */ + @Synchronized + private fun registerSession( + sessionId: String, + session: SessionProcess, + ) { + sessions[sessionId] = session + hasLiveSessions.value = true + } + + /** + * The only place [sessions] loses an entry. With [onlyProcess] set, an entry that has since been + * replaced by a newer process is left alone - the case of an old reader job finishing after its + * session was relaunched. Returns the removed entry, or null if nothing was removed. + */ + @Synchronized + private fun unregisterSession( + sessionId: String, + onlyProcess: Process? = null, + ): SessionProcess? { + val session = sessions[sessionId] ?: return null + if (onlyProcess != null && session.process !== onlyProcess) return null + sessions.remove(sessionId) + hasLiveSessions.value = sessions.isNotEmpty() + return session + } + + /** + * Starts the watcher that turns permission requests parked on disk into events. It is a single + * long-lived coroutine, but it only polls while a CLI process exists: the hook that parks a + * request dies with its process (see [pendingQuestions]), so with none alive there is nothing + * that could appear. Without that it kept reading the directory every [BRIDGE_POLL_INTERVAL_MILLIS] + * for the rest of the process once any session had ever started. + */ private fun ensureBridgeWatcher() { if (bridgeWatchJob?.isActive == true) return bridgeWatchJob = scope.launch { - while (isActive) { - permissionBridge.pollPending().forEach { request -> - when (request.kind) { - ClaudePermissionBridge.Kind.QUESTION -> { - val question = permissionBridge.toQuestionRequest(request) - if (question != null) { - events.tryEmit(OpenCodeEvent.QuestionAsked(question)) - } else { - events.tryEmit( - OpenCodeEvent.PermissionAsked(permissionBridge.toPermissionRequest(request)), - ) - } - } - else -> - events.tryEmit( - OpenCodeEvent.PermissionAsked(permissionBridge.toPermissionRequest(request)), - ) - } + hasLiveSessions.pollWhileActive(BRIDGE_POLL_INTERVAL_MILLIS) { emitPendingBridgeRequests() } + } + } + + private fun emitPendingBridgeRequests() { + permissionBridge.pollPending().forEach { request -> + when (request.kind) { + ClaudePermissionBridge.Kind.QUESTION -> { + val question = permissionBridge.toQuestionRequest(request) + if (question != null) { + events.tryEmit(OpenCodeEvent.QuestionAsked(question)) + } else { + events.tryEmit( + OpenCodeEvent.PermissionAsked(permissionBridge.toPermissionRequest(request)), + ) } - delay(250) } + else -> + events.tryEmit( + OpenCodeEvent.PermissionAsked(permissionBridge.toPermissionRequest(request)), + ) } + } } fun update() { @@ -300,7 +344,7 @@ class ClaudeCodeRuntime( /** Stops the process backing [sessionId]; the next send resumes the same Claude conversation. */ @Synchronized fun stop(sessionId: String) { - sessions.remove(sessionId)?.let { session -> + unregisterSession(sessionId)?.let { session -> session.readerJob.cancel() if (session.process.isAlive) session.process.destroyForcibly() } @@ -398,9 +442,7 @@ class ClaudeCodeRuntime( }.exceptionOrNull() .takeUnless { it is CancellationException } messageStore.flush() - synchronized(this@ClaudeCodeRuntime) { - if (sessions[sessionId]?.process === process) sessions.remove(sessionId) - } + unregisterSession(sessionId, onlyProcess = process) // stop() cancels this job; the abort path owns the state transitions, and an idle // emitted here would announce the killed run as completed. if (!isActive) return@launch @@ -417,7 +459,7 @@ class ClaudeCodeRuntime( } return SessionProcess(process, readerJob, parser, permissionMode, directory, model, effort, systemPrompt) - .also { sessions[sessionId] = it } + .also { registerSession(sessionId, it) } } private fun processArguments( @@ -768,5 +810,8 @@ class ClaudeCodeRuntime( const val TITLE_TIMEOUT_SECONDS = 45L const val WORKSPACE_COMMAND_TIMEOUT_SECONDS = 60L const val INTERRUPTED_TOOL_ERROR = "Claude Code session ended before the tool result was received" + + /** How often a live session's parked permission requests are looked for. */ + const val BRIDGE_POLL_INTERVAL_MILLIS = 250L } } diff --git a/app/src/main/java/com/yugahashimoto/andcode/ui/AndCodeApp.kt b/app/src/main/java/com/yugahashimoto/andcode/ui/AndCodeApp.kt index ca67efde7..c0ca2ca71 100644 --- a/app/src/main/java/com/yugahashimoto/andcode/ui/AndCodeApp.kt +++ b/app/src/main/java/com/yugahashimoto/andcode/ui/AndCodeApp.kt @@ -711,6 +711,7 @@ fun AndCodeApp( workspacePath = chatState.sessionDirectory ?: chatState.selectedWorkspacePath, agent = selectedRuntime?.agent, runtimeDirectory = java.io.File(context.filesDir, "runtime"), + foreground = app.appForeground.foreground, onOpenUrl = { url -> navController.navigate(guestBrowserRoute(url)) { launchSingleTop = true } }, diff --git a/app/src/test/java/com/yugahashimoto/andcode/core/util/PollWhileActiveTest.kt b/app/src/test/java/com/yugahashimoto/andcode/core/util/PollWhileActiveTest.kt new file mode 100644 index 000000000..231c816e8 --- /dev/null +++ b/app/src/test/java/com/yugahashimoto/andcode/core/util/PollWhileActiveTest.kt @@ -0,0 +1,113 @@ +package com.yugahashimoto.andcode.core.util + +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Test + +private const val INTERVAL_MILLIS = 250L + +/** + * Covers the gate every always-on file poll in the app now sits behind: a loop that has nothing to + * watch for must be a suspended coroutine, not a timer that keeps waking the process every few + * hundred milliseconds for the rest of its life. + */ +@OptIn(ExperimentalCoroutinesApi::class) +class PollWhileActiveTest { + @Test + fun `never polls while the gate stays closed`() = + runTest { + val scope = TestScope(StandardTestDispatcher(testScheduler)) + val gate = MutableStateFlow(false) + var polls = 0 + try { + scope.launch { gate.pollWhileActive(INTERVAL_MILLIS) { polls++ } } + advanceTimeBy(100 * INTERVAL_MILLIS) + runCurrent() + + assertEquals(0, polls) + } finally { + scope.cancel() + } + } + + @Test + fun `polls immediately when the gate is open and then once per interval`() = + runTest { + val scope = TestScope(StandardTestDispatcher(testScheduler)) + val gate = MutableStateFlow(true) + var polls = 0 + try { + scope.launch { gate.pollWhileActive(INTERVAL_MILLIS) { polls++ } } + runCurrent() + assertEquals(1, polls) + + advanceTimeBy(INTERVAL_MILLIS) + runCurrent() + assertEquals(2, polls) + + advanceTimeBy(INTERVAL_MILLIS) + runCurrent() + assertEquals(3, polls) + } finally { + scope.cancel() + } + } + + @Test + fun `stops polling as soon as the gate closes`() = + runTest { + val scope = TestScope(StandardTestDispatcher(testScheduler)) + val gate = MutableStateFlow(true) + var polls = 0 + try { + scope.launch { gate.pollWhileActive(INTERVAL_MILLIS) { polls++ } } + runCurrent() + assertEquals(1, polls) + + gate.value = false + runCurrent() + advanceTimeBy(100 * INTERVAL_MILLIS) + runCurrent() + + assertEquals(1, polls) + } finally { + scope.cancel() + } + } + + /** + * Returning to the app has to pick up whatever arrived while it was away, so a reopened gate + * polls at once instead of first waiting out an interval. + */ + @Test + fun `polls straight away when the gate reopens`() = + runTest { + val scope = TestScope(StandardTestDispatcher(testScheduler)) + val gate = MutableStateFlow(true) + var polls = 0 + try { + scope.launch { gate.pollWhileActive(INTERVAL_MILLIS) { polls++ } } + runCurrent() + gate.value = false + runCurrent() + advanceTimeBy(10 * INTERVAL_MILLIS) + runCurrent() + assertEquals(1, polls) + + gate.value = true + runCurrent() + + assertEquals(2, polls) + } finally { + scope.cancel() + } + } +} diff --git a/app/src/test/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcherTest.kt b/app/src/test/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcherTest.kt index 2cafaa275..e6365313a 100644 --- a/app/src/test/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcherTest.kt +++ b/app/src/test/java/com/yugahashimoto/andcode/feature/browser/BrowserCommandWatcherTest.kt @@ -2,6 +2,15 @@ package com.yugahashimoto.andcode.feature.browser import com.yugahashimoto.andcode.core.storage.DeviceStorage import com.yugahashimoto.andcode.runtime.LocalAgent +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNull @@ -125,6 +134,40 @@ class BrowserCommandWatcherTest { assertTrue(file.exists()) } + /** + * The guest browser is only worth opening for someone looking at the app, so the watcher stays + * parked while it is in the background and hands over what arrived the moment it returns. + */ + @OptIn(ExperimentalCoroutinesApi::class) + @Test + fun `watcher opens a pending command only while the app is in the foreground`() = + runTest { + val foreground = MutableStateFlow(false) + val opened = mutableListOf() + var pending: String? = "https://www.google.com/" + val scope = TestScope(StandardTestDispatcher(testScheduler)) + try { + scope.launch { + watchBrowserCommands( + active = foreground, + intervalMillis = 1_000L, + poll = { pending.also { pending = null } }, + onOpenUrl = { opened += it }, + ) + } + advanceTimeBy(60_000L) + runCurrent() + assertEquals(emptyList(), opened) + + foreground.value = true + runCurrent() + + assertEquals(listOf("https://www.google.com/"), opened) + } finally { + scope.cancel() + } + } + private fun consume( path: String?, agent: LocalAgent? = LocalAgent.CODEX, diff --git a/app/src/test/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridgeTest.kt b/app/src/test/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridgeTest.kt index 56c38b74b..16f551182 100644 --- a/app/src/test/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridgeTest.kt +++ b/app/src/test/java/com/yugahashimoto/andcode/feature/schedule/ScheduleBridgeTest.kt @@ -3,6 +3,15 @@ package com.yugahashimoto.andcode.feature.schedule import com.yugahashimoto.andcode.data.schedule.Schedule import com.yugahashimoto.andcode.data.schedule.ScheduleRun import com.yugahashimoto.andcode.data.schedule.ScheduleRunStatus +import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest import org.json.JSONObject import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse @@ -77,6 +86,7 @@ class ScheduleBridgeTest { requestId: String, op: String, args: JSONObject = JSONObject(), + createdAtMs: Long = 0L, ) { val file = File(pendingDir(bridge), "$requestId.json") file.parentFile?.mkdirs() @@ -84,7 +94,7 @@ class ScheduleBridgeTest { JSONObject() .put("op", op) .put("args", args) - .put("createdAtMs", 0L) + .put("createdAtMs", createdAtMs) .toString(), ) } @@ -442,4 +452,107 @@ class ScheduleBridgeTest { assertTrue(!orphan.exists()) } + + /** + * Nothing can be waiting on the bridge while the app is backgrounded and no work is in flight - + * the device is free to suspend and the guest freezes with it - so the loop must stay parked + * then, and answer at once when the gate opens again. + */ + @OptIn(ExperimentalCoroutinesApi::class) + @Test + fun `run answers pending requests only while the gate is open`() = + runTest { + val bridge = bridgeWith(FakeStore()) + val id = requestId() + writeRequest(bridge, id, "list") + val gate = MutableStateFlow(false) + val scope = TestScope(StandardTestDispatcher(testScheduler)) + try { + scope.launch { bridge.run(gate) } + advanceTimeBy(10 * ScheduleBridge.POLL_INTERVAL_MILLIS) + runCurrent() + assertFalse(File(responsesDir(bridge), "$id.json").exists()) + + gate.value = true + runCurrent() + + assertTrue(File(responsesDir(bridge), "$id.json").exists()) + } finally { + scope.cancel() + } + } + + /** + * The guest stops waiting after 90 seconds and tells the agent the call failed. A request that + * sat unanswered past that - written while the gate was shut - must not run on the way back in, + * or a `create`/`delete`/`runNow` the agent was told did not happen would happen anyway. + */ + @Test + fun `a request the guest has already given up on is dropped with an error instead of run`() { + val store = FakeStore() + val now = System.currentTimeMillis() + val bridge = bridgeWith(store, clock = { now }) + val id = requestId() + writeRequest(bridge, id, "create", createArgs(), createdAtMs = now - 5 * 60_000L) + + bridge.pollOnce() + + val response = readResponse(bridge, id) + assertFalse(response.getBoolean("ok")) + assertTrue(response.getString("error").contains("expired")) + assertEquals(0, store.schedules.size) + assertFalse(File(pendingDir(bridge), "$id.json").exists()) + } + + @Test + fun `a request the guest is still waiting on is run`() { + val store = FakeStore() + val now = System.currentTimeMillis() + val bridge = bridgeWith(store, clock = { now }) + val id = requestId() + writeRequest(bridge, id, "create", createArgs(), createdAtMs = now - 30_000L) + + bridge.pollOnce() + + assertTrue(readResponse(bridge, id).getBoolean("ok")) + assertEquals(1, store.schedules.size) + } + + @Test + fun `the expiry line falls at the configured age`() { + val store = FakeStore() + val now = System.currentTimeMillis() + val bridge = bridgeWith(store, clock = { now }) + val inside = requestId() + val outside = requestId() + val limit = ScheduleBridge.REQUEST_EXPIRES_AFTER_MILLIS + writeRequest(bridge, inside, "create", createArgs(prompt = "inside"), createdAtMs = now - limit + 1_000L) + writeRequest(bridge, outside, "create", createArgs(prompt = "outside"), createdAtMs = now - limit - 1_000L) + + bridge.pollOnce() + + assertTrue(readResponse(bridge, inside).getBoolean("ok")) + assertFalse(readResponse(bridge, outside).getBoolean("ok")) + assertEquals(listOf("inside"), store.schedules.map { it.prompt }) + } + + /** Only a timestamp that is plainly in the past counts as old; anything else is run as before. */ + @Test + fun `a request stamped in the future or with no timestamp is run`() { + val store = FakeStore() + val now = System.currentTimeMillis() + val bridge = bridgeWith(store, clock = { now }) + val future = requestId() + val missing = requestId() + writeRequest(bridge, future, "create", createArgs(prompt = "future"), createdAtMs = now + 60_000L) + File(pendingDir(bridge), "$missing.json").writeText( + JSONObject().put("op", "create").put("args", createArgs(prompt = "missing")).toString(), + ) + + bridge.pollOnce() + + assertTrue(readResponse(bridge, future).getBoolean("ok")) + assertTrue(readResponse(bridge, missing).getBoolean("ok")) + assertEquals(setOf("future", "missing"), store.schedules.map { it.prompt }.toSet()) + } } diff --git a/docs/LOCAL_RUNTIME.md b/docs/LOCAL_RUNTIME.md index e19646de7..716179f2f 100644 --- a/docs/LOCAL_RUNTIME.md +++ b/docs/LOCAL_RUNTIME.md @@ -218,6 +218,7 @@ OpenCodeがOAuthの `auto` 方式を返した場合、Androidはブラウザを - `andcode-schedule-mcp.py` が `/usr/local/bin` に入り、全エージェントのMCP設定（OpenCodeの`opencode.json`、Claude Codeの`.claude.json`、Antigravityの`mcp_config.json`）へ `and-code-schedule` サーバーとして登録される - サーバーは `/workspace/.and-code/schedule-bridge/pending/.json` へリクエストを書き、アプリ側の`ScheduleBridge`がポーリングして実行し、`responses/.json` へ結果を書き戻す +- ポーリングは、アプリが前面にあるか、ランタイムが作業中（チャットのセッション実行・スケジュール実行など）の間だけ動く。どちらでもないときは端末がスリープでき、ゲストも止まっているため待機し、前面に戻るか作業が始まった時点で即座に処理する。ゲスト側の応答待ち（90秒）を過ぎた古いリクエストは、実行せず期限切れエラーで破棄する - 書き込み系はスケジューリングUIと同じく暗号化リポジトリとアラーム（`ScheduleManager`）を通るため、変更はアプリのスケジュール画面へ即反映される - 提供ツール: `schedule_list` / `schedule_get` / `schedule_runs` / `schedule_create` / `schedule_update` / `schedule_delete` / `schedule_set_enabled` / `schedule_run_now`