Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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()
}
Expand Down
33 changes: 33 additions & 0 deletions app/src/main/java/com/yugahashimoto/andcode/core/util/FlowExt.kt
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -79,3 +81,34 @@ fun Flow<Boolean>.debounceFalseEdge(graceMillis: Long): Flow<Boolean> =
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<Boolean>.pollWhileActive(
intervalMillis: Long,
action: suspend () -> Unit,
) {
collectLatest { active ->
if (!active) return@collectLatest
while (true) {
action()
delay(intervalMillis)
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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<Boolean>,
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<Boolean>,
poll: suspend () -> String?,
onOpenUrl: (String) -> Unit,
intervalMillis: Long = POLL_INTERVAL_MILLIS,
) {
active.pollWhileActive(intervalMillis) { poll()?.let(onOpenUrl) }
}
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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<Boolean>) {
// 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. */
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -78,6 +86,7 @@ class ClaudeCodeRuntime(
val systemPrompt: String?,
)

/** Mutate only through [registerSession] and [unregisterSession], which keep [hasLiveSessions] in step. */
private val sessions = linkedMapOf<String, SessionProcess>()

/**
Expand Down Expand Up @@ -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() {
Expand Down Expand Up @@ -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()
}
Expand Down Expand Up @@ -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
Expand All @@ -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(
Expand Down Expand Up @@ -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
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
},
Expand Down
Loading
Loading