diff --git a/AGENTS.md b/AGENTS.md index 448e210..81ca558 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -209,6 +209,22 @@ first if a remote spawn ever mangles an argument. word rather than a bare spinner because "deleting" and "importing" differ in kind, and the inertness is the overlay consuming pointer events rather than each caller remembering to disable its own click handler. +- **Importing and deleting run on the server, not in the request.** `DELETE + /setups/{id}/importable/{session}` and `POST + /setups/{id}/importable/{session}/import` both answer 202 and do the work + in a spawned task, because the phone that asked is free to leave and used + to cancel its own batch by doing so. What replaces the reply is + `session::pending`: every row of the listing carries `pending` and + `error`, and `GET /setups/{id}/importable/events` streams the changes. + **Both, not either.** The stream is a broadcast with no memory, so an + operation that starts and finishes while it is still connecting is one + nothing will ever be said about -- that left a row marked "waiting" for + ever, and the listing is what repairs it. So the screen fetches again + after a handover when anything still looks outstanding, and takes the row + states from the answer rather than from what it remembers. +- **A single tap still waits.** "Continue this and take me to it" needs the + session it made, and 202 does not carry one. The batch and the tap share + `spawn` on the server so the two cannot drift about what importing means. - **The import screen selects in batches: hold to enter, tap to add.** The options that act on a selection appear along the bottom, and are Delete and Import only. Submitting clears the selection immediately and marks @@ -271,7 +287,12 @@ first if a remote spawn ever mangles an argument. real `--resume` on the owner's account. Neither is a price worth paying to look at a list. It shares the real TLS certificates, because the installed APK pins that CA, so run it while the ordinary server is down. - It passes `--delay` by default for the reason the next entry gives. + It passes `--delay` by default for the reason the next entry gives, and + `AI_SANDBOX_BIG_MB` puts one large transcript among the small ones -- + `AI_SANDBOX_SPAWN_DELAY` makes the fake CLI slow to start. Both exist + because operations that finish in milliseconds have states on the way that + nothing can observe, and an unobservable state is one where broken and + working look identical. - **`ai-server --delay MS` holds every response back.** Over the tunnel a phone's requests take tens to hundreds of milliseconds, and several faults live entirely in what the app does *while* one is outstanding. On @@ -522,6 +543,14 @@ machine belongs in `~/.claude/TOOLCHAIN.md` (toolchain versions) or perfectly -- while the usage dialog, reading the same field, quietly drew nothing. `WindowEnd` in `ResetCountdown.kt` is now the one rule both go through. +- **Resolving one importable session used to list every one of them.** + `import::delete` and the import seed both called `list`, which reads every + transcript Claude Code has ever written -- measured at 3.7 seconds against + the 867 MB in this VM, paid once per session in a batch. `import::find` + takes the same script with one glob narrower, and `delete` resolves the + path itself: 78ms. Ids are checked (`is_session_id`) before they reach + that glob, since a `/` or `..` in one walks it out of the projects + directory and `delete` removes what it lands on. - **A transcript page used to cost the whole transcript.** `read_window` read and parsed every line and then kept the last `limit` of them, so the work was the size of the conversation rather than the size of the answer: diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt index a496e19..65d5287 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -283,8 +283,50 @@ data class Importable( * server refuses an import of a "yes"; the row says so before you press it. */ val inUse: String, + /** + * What this server is doing to the session right now -- "importing" or "deleting" -- or null + * when nothing is. + * + * The server's answer rather than the phone's, because the work outlives the screen that asked + * for it: leaving the import list and coming back has to show what is still running, and a + * phone that was asleep or out of range never saw the events that said so. + */ + val pending: String?, + /** + * How the last attempt on this row failed, if it did. Kept by the server until something + * replaces it, for the same reason [pending] is the server's to answer. + */ + val error: String?, ) +/** + * One frame of `GET /setups/{id}/importable/events`: an operation starting, finishing or failing. + * + * [operation] is only set by a start, and [message] only by a failure -- the three states are every + * way an operation can be, and each carries exactly what that state knows. + */ +data class ImportableChange( + val session: String, + val state: String, + val operation: String?, + val message: String?, +) + +fun parseImportableChange(payload: String): ImportableChange? = + try { + val frame = JSONObject(payload) + ImportableChange( + session = frame.getString("session"), + state = frame.getString("state"), + operation = frame.optString("operation").takeIf { it.isNotEmpty() }, + message = frame.optString("message").takeIf { it.isNotEmpty() }, + ) + } catch (_: org.json.JSONException) { + // A frame this build does not understand is not a reason to drop the stream: the listing + // is the truth and will say what happened whatever this missed. + null + } + /** * What a machine has that could be continued. * @@ -314,6 +356,8 @@ fun fetchImportable(settings: ServerSettings, setup: String): List = // "unknown" says -- so the default is the honest one rather than "no". inUse = session.optString("inUse", "unknown"), named = session.optBoolean("named", false), + pending = session.optString("pending").takeIf { it.isNotEmpty() }, + error = session.optString("error").takeIf { it.isNotEmpty() }, ) } } @@ -590,10 +634,47 @@ fun startSession(settings: ServerSettings, sessionId: String) { * The transcript *is* the session, so this ends any chance of resuming that conversation. The * caller confirms first; see ImportScreen. */ +/** + * Asks the machine to delete a Claude Code session, and returns as soon as it has accepted. + * + * The work runs on the server, so this returning is not the same as it being done -- what says that + * is the row's own state, through [fetchImportable] and the change stream. That is the point: + * leaving the screen used to cancel the delete it had started. + */ fun deleteImportable(settings: ServerSettings, setup: String, sessionId: String) { requestFromServer(settings, "/setups/$setup/importable/$sessionId", method = "DELETE") {} } +/** + * Continues a Claude Code session in the background, returning once the server has accepted. + * + * Separate from [spawnSession] because the two are asked different questions. That one means "start + * this and take me to it", so it waits and answers with the session. This is the import list's + * batch: several at once, nobody waiting on any particular one, and the result arrives as a row + * changing rather than as a reply -- which is what lets the screen be left. + */ +fun startImport( + settings: ServerSettings, + setup: String, + sessionId: String, + provider: String, + permissionMode: String? = null, + model: String? = null, +) { + val body = + JSONObject().apply { + put("provider", provider) + permissionMode?.let { put("permissionMode", it) } + model?.let { put("model", it) } + } + requestFromServer( + settings, + "/setups/$setup/importable/$sessionId/import", + method = "POST", + jsonBody = body.toString(), + ) {} +} + /** * A page of a session's transcript, oldest first within the page. * diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt index 20749d3..c031e1a 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/EventStream.kt @@ -1,9 +1,5 @@ package com.example.aiapp -import java.io.IOException -import java.net.HttpURLConnection -import java.net.URL - /** * The frame name the server uses to say a cursor was too far behind to continue from. Must match * `send_backlog` in the backend's routes.rs. @@ -14,88 +10,29 @@ private const val RESET_EVENT = "reset" * The SSE half of the API: one long-lived GET per open session screen, replaying the transcript * after a cursor and then following it live. * - * Blocking -- run() occupies its thread until the stream ends. [close] (from any thread) is the - * cancellation path: it disconnects the socket, which unblocks the read; run() then returns instead - * of throwing, so a deliberate close doesn't surface as a connection error. The caller owns - * reconnecting (with the last seq it saw as the new cursor) -- see SessionScreen. + * The connection and its framing belong to [Sse]; what stays here is what this stream's frames + * mean. [close] from any thread ends it, and the caller owns reconnecting -- with the last seq it + * saw as the new cursor. See SessionScreen. */ -class EventStream(private val settings: ServerSettings, private val sessionId: String) { - @Volatile private var connection: HttpURLConnection? = null - @Volatile private var closed = false +class EventStream(settings: ServerSettings, private val sessionId: String) { + private val stream = Sse(settings) - fun close() { - closed = true - connection?.disconnect() - } + fun close() = stream.close() /** * Streams events after [after] into [onEvent] until the stream drops. * - * [onOpen] fires once the server has accepted the connection. That is the measured moment the - * stream is live again, and the only honest thing to clear a previous failure on: an earlier - * version cleared on the first event instead, so an idle session went on displaying a - * connection error that had already been recovered from, indefinitely. - * * [onReset] fires when the server answers that the cursor is too far behind to continue from: * everything already displayed is stale and the events that follow are a fresh window, so the * caller drops what it holds and rebuilds -- the same thing it does when the screen opens. It * arrives before those events, so a caller that clears on it stays in order. */ fun run(after: Long, onOpen: () -> Unit, onReset: () -> Unit, onEvent: (SeqEvent) -> Unit) { - val connection = - URL("${settings.baseUrl}/sessions/$sessionId/events?after=$after").openConnection() - as HttpURLConnection - this.connection = connection - try { - connection.applyPinnedTls() - connection.connectTimeout = CONNECT_TIMEOUT_MS - // No read timeout: between events there is nothing to read for - // as long as the session is idle; the server's keep-alives and - // a dead socket erroring out are the liveness story. - connection.readTimeout = 0 - connection.setRequestProperty("Authorization", "Bearer ${settings.token}") - connection.setRequestProperty("Accept", "text/event-stream") - if (connection.responseCode != 200) { - val detail = connection.errorStream?.bufferedReader()?.readText()?.trim() - throw ApiException(detail ?: "HTTP ${connection.responseCode} for the event stream") - } - - onOpen() - val reader = connection.inputStream.bufferedReader() - // SSE framing: `data:` and `event:` lines accumulate until a - // blank line ends the frame. `id:` (the seq) is also inside the - // JSON payload, so it needs no separate handling; comment lines - // (keep-alives) start with ':' and are skipped. - val data = StringBuilder() - var name: String? = null - while (true) { - val line = reader.readLine() ?: break - when { - line.isEmpty() -> { - // A named frame carries no payload and a data frame - // has no name, so this is one or the other. - if (name == RESET_EVENT) onReset() - else if (data.isNotEmpty()) onEvent(parseSeqEvent(data.toString())) - data.clear() - name = null - } - line.startsWith("data:") -> data.append(line.removePrefix("data:").trim()) - line.startsWith("event:") -> name = line.removePrefix("event:").trim() - else -> {} // id:, comments -- nothing to do - } - } - } catch (e: ApiException) { - throw e - } catch (e: IOException) { - if (!closed) { - throw ApiException( - "Can't reach the server -- retrying. (${e.message ?: e::class.simpleName})", - e, - ) - } - } finally { - connection.disconnect() - this.connection = null + stream.run("/sessions/$sessionId/events?after=$after", onOpen) { name, data -> + // A named frame carries no payload and a data frame has no name, so this is one or + // the other. + if (name == RESET_EVENT) onReset() + else if (data.isNotEmpty()) onEvent(parseSeqEvent(data)) } } } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt index a2101fd..7e50612 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportScreen.kt @@ -24,6 +24,7 @@ import androidx.compose.material3.Surface import androidx.compose.material3.Text import androidx.compose.material3.TextButton import androidx.compose.runtime.Composable +import androidx.compose.runtime.DisposableEffect import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateOf @@ -38,6 +39,9 @@ import androidx.compose.ui.text.style.TextOverflow import androidx.compose.ui.unit.Dp import androidx.compose.ui.unit.dp import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.delay import kotlinx.coroutines.launch import kotlinx.coroutines.withContext @@ -115,19 +119,36 @@ fun ImportScreen(settings: ServerSettings, reloadToken: Int, onImported: (Sessio val movedAt = remember { mutableMapOf() } fun settling(id: String) = System.currentTimeMillis() - (movedAt[id] ?: 0L) < SETTLE_MS + /** + * Fetches the list and takes the row states from it. + * + * Taken from the answer rather than kept across the load: the server is what knows what is + * running, and this screen may be opening on work another screen -- or another phone -- + * started. Anything held locally would be a second version of that, and the stale one. + */ + suspend fun fetchInto(setup: Setup): LoadState> = + try { + val rows = withContext(Dispatchers.IO) { fetchImportable(settings, setup.id) } + running = rows.mapNotNull { row -> row.pending?.let { row.id to it } }.toMap() + rowErrors = rows.mapNotNull { row -> row.error?.let { row.id to it } }.toMap() + LoadState.Loaded(rows) + } catch (err: Exception) { + LoadState.Error(err.message ?: "Couldn't list sessions") + } + fun loadSessions(setup: Setup) { sessions = LoadState.Loading selected = emptySet() - rowErrors = emptyMap() - scope.launch { - sessions = - try { - LoadState.Loaded( - withContext(Dispatchers.IO) { fetchImportable(settings, setup.id) } - ) - } catch (err: Exception) { - LoadState.Error(err.message ?: "Couldn't list sessions") - } + scope.launch { sessions = fetchInto(setup) } + } + + /** Takes a row out of the list, once the machine no longer has it to offer. */ + fun forget(id: String) { + val loaded = sessions + if (loaded is LoadState.Loaded) { + // Only this row, and only what changed -- refetching instead put every other row back + // through a loading spinner to report a change that was never in doubt. + sessions = LoadState.Loaded(loaded.value.filterNot { it.id == id }) } } @@ -146,58 +167,62 @@ fun ImportScreen(settings: ServerSettings, reloadToken: Int, onImported: (Sessio } /** - * Runs [operation] over [targets] one at a time, marking each row with [label] while its turn - * lasts and taking it off the list when it succeeds. + * Hands [targets] to the server, marking each row as it goes. * - * One runner for both operations and for both the single tap and the batch, so "what a row - * looks like while something is happening to it" and "what happens when one of ten fails" are - * decided once. Sequentially, because each import starts a CLI on the machine and ten at once - * is a load nobody asked for; the reader sees the work walk down the list, which is also the - * only honest progress this screen can show. + * The requests only *start* the work now -- the server runs it and says how it went on the + * change stream, which is what lets this screen be left while a batch is still going. So there + * is nothing here to wait for and nothing to sequence: each row is marked, its request goes, + * and everything after that arrives as an event. * - * The selection is dropped the moment the work is handed over, not when it finishes: the screen - * goes back to how it started, and what says the work is happening is the rows it is happening - * to. Holding the selection until the end left the bar up over rows that could no longer be - * pressed, offering to start again something already running. + * Marked [WAITING] rather than with the operation's own word until the server confirms. Between + * the request leaving and the `started` event coming back, "we have asked" is the truth and "it + * is importing" is a guess -- and the row is inert either way, which is the part that matters. * - * A failure keeps its row and puts the server's words on it. Selecting those rows again is then - * the reader's decision rather than a state the screen carried for them — and it is the - * decision worth making deliberately, because retrying a delete that the server refused is - * usually not what somebody wants to do by pressing the same button twice. + * The selection is dropped as the work is handed over, not when it finishes: the screen goes + * back to how it started, and what says the work is happening is the rows it is happening to. */ - fun runOn(targets: List, label: String, operation: suspend (Importable) -> Unit) { + fun handOver(targets: List, send: suspend (Importable) -> Unit) { selected = emptySet() running = running + targets.associate { it.id to WAITING } + rowErrors = rowErrors - targets.map { it.id }.toSet() + val setup = chosen scope.launch { - for (target in targets) { - running = running + (target.id to label) - rowErrors = rowErrors - target.id - try { - operation(target) - val loaded = sessions - if (loaded is LoadState.Loaded) { - // As each one lands, not all of them at the end. Holding the finished - // rows in place to keep the list still was tried and is worse: a row - // that has been imported but is still sitting there looks exactly like - // one that has not, and tapping it starts a second CLI on the same - // transcript. A row that is gone cannot be tapped at all. - // - // Only this row, and only what changed -- refetching instead put every - // other row back through a loading spinner to report a change that was - // never in doubt. - val now = System.currentTimeMillis() - loaded.value - .asSequence() - .dropWhile { it.id != target.id } - .drop(1) - .forEach { movedAt[it.id] = now } - sessions = LoadState.Loaded(loaded.value.filterNot { it.id == target.id }) + // All at once rather than a loop that awaits each: the handover is what has to + // survive leaving the screen, so it should take one round trip rather than one per + // row. What each request starts is already safe once the server has it. + targets + .map { target -> + async { + try { + withContext(Dispatchers.IO) { send(target) } + } catch (err: Exception) { + // The server never took it, so nothing is running and no event will + // arrive to say so. This is the one failure the screen must report + // itself. + running = running - target.id + rowErrors = rowErrors + (target.id to (err.message ?: "Couldn't ask")) + } } - } catch (err: Exception) { - rowErrors = rowErrors + (target.id to (err.message ?: "Didn't work")) - } finally { - running = running - target.id } + .awaitAll() + + // Then ask what actually happened, if anything still looks outstanding. + // + // The change stream is a broadcast with no memory, so an operation that started and + // finished while it was still connecting is one nothing will ever be said about -- + // and the row sits marked for ever. That is not hypothetical: with responses held + // back far enough for the stream to open late, one row of a pair of deletes cleared + // and the other stayed on "waiting". + // + // The listing is the repair, because it carries the same state the events do. Only + // when something still looks outstanding, so the ordinary case -- where the events + // arrived and the rows are already gone -- does not pay for a second listing, which + // is the most expensive call this screen makes. + if (setup != null && targets.any { running.containsKey(it.id) }) { + // Quietly: no Loading, because blanking the list to report on rows that are + // already saying what is happening to them is the flicker this screen avoids + // everywhere else. + sessions = fetchInto(setup) } } } @@ -211,26 +236,111 @@ fun ImportScreen(settings: ServerSettings, reloadToken: Int, onImported: (Sessio * and then looking at it is what a tap on a row means, and a batch has several results and no * reason to pick one of them to become the screen. */ - fun importAll(targets: List, thenOpen: Boolean) { + /** Continues [targets] in the background, leaving the screen where it is. */ + fun importAll(targets: List) { val setup = chosen ?: return val useProvider = provider ?: return - runOn(targets, IMPORTING) { session -> - val spawned = - withContext(Dispatchers.IO) { - spawnSession( - settings, - setup = setup.id, - provider = useProvider.name, - // Nothing to say: the server titles it from the session it is continuing. - title = "", - permissionMode = permissionMode, - import = session.id, - ) - } - if (thenOpen) onImported(spawned) + handOver(targets) { session -> + startImport( + settings, + setup = setup.id, + sessionId = session.id, + provider = useProvider.name, + permissionMode = permissionMode, + ) } } + /** + * Continues one session and goes to it. + * + * The tap keeps waiting, because "take me there" needs the session it made and the server's + * accepted-and-running answer does not carry one. It is one session and somebody is watching + * it, which is the case where waiting is the right thing anyway. + */ + fun importAndOpen(target: Importable) { + val setup = chosen ?: return + val useProvider = provider ?: return + running = running + (target.id to IMPORTING) + rowErrors = rowErrors - target.id + scope.launch { + try { + val spawned = + withContext(Dispatchers.IO) { + spawnSession( + settings, + setup = setup.id, + provider = useProvider.name, + // Nothing to say: the server titles it from the session it continues. + title = "", + permissionMode = permissionMode, + import = target.id, + ) + } + forget(target.id) + onImported(spawned) + } catch (err: Exception) { + rowErrors = rowErrors + (target.id to (err.message ?: "Couldn't import that one")) + } finally { + running = running - target.id + } + } + } + + // Live changes to what the server is doing to these sessions, for as long as this screen is + // up. The listing already carried the same state when the screen opened -- this is what keeps + // it current afterwards, including for work another screen or another phone started. + // + // Failures here are deliberately quiet. There is nothing for a reader to do about a dropped + // event stream, and nothing is lost by one: every state it would have carried is in the next + // listing, which is what Refresh and re-entering the tab already fetch. + val liveChanges = remember { + java.util.concurrent.atomic.AtomicReference(null) + } + LaunchedEffect(chosen?.id) { + val setup = chosen?.id ?: return@LaunchedEffect + try { + while (true) { + val stream = ImportableStream(settings, setup) + liveChanges.set(stream) + try { + withContext(Dispatchers.IO) { + stream.run(onOpen = {}) { change -> + when (change.state) { + "started" -> + running = + running + (change.session to (change.operation ?: WAITING)) + // Gone from the machine either way: a delete removed the + // transcript, an import made it a session, and neither is + // something this list still has to offer. + "finished" -> { + running = running - change.session + forget(change.session) + } + "failed" -> { + running = running - change.session + rowErrors = + rowErrors + + (change.session to (change.message ?: "Didn't work")) + } + } + } + } + } catch (_: ApiException) { + // Retried below; the listing is the truth in the meantime. + } finally { + stream.close() + } + delay(RECONNECT_DELAY_MS) + } + } finally { + // Cancellation cannot interrupt a blocking socket read; closing is what unblocks it. + liveChanges.getAndSet(null)?.close() + } + } + // The screen leaving the composition entirely, which the effect above does not cover. + DisposableEffect(chosen?.id) { onDispose { liveChanges.get()?.close() } } + // Back leaves selection mode rather than the tab, which is the level it is one step above. // Nested inside MainScreen's own handler, so it wins while there is a selection. BackHandler(enabled = selected.isNotEmpty()) { selected = emptySet() } @@ -305,7 +415,7 @@ fun ImportScreen(settings: ServerSettings, reloadToken: Int, onImported: (Sessio if (session.id in selected) selected - session.id else selected + session.id }, - onOpen = { session -> importAll(listOf(session), thenOpen = true) }, + onOpen = { session -> importAndOpen(session) }, ) } } @@ -324,7 +434,7 @@ fun ImportScreen(settings: ServerSettings, reloadToken: Int, onImported: (Sessio barHeight = with(density) { it.height.toDp() } }, onDelete = { confirming = picked }, - onImport = { importAll(picked, thenOpen = false) }, + onImport = { importAll(picked) }, ) } } @@ -354,10 +464,8 @@ fun ImportScreen(settings: ServerSettings, reloadToken: Int, onImported: (Sessio onClick = { val setup = chosen ?: return@TextButton confirming = null - runOn(targets, DELETING) { session -> - withContext(Dispatchers.IO) { - deleteImportable(settings, setup.id, session.id) - } + handOver(targets) { session -> + deleteImportable(settings, setup.id, session.id) } } ) { diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/ImportableStream.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportableStream.kt new file mode 100644 index 0000000..4009a54 --- /dev/null +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/ImportableStream.kt @@ -0,0 +1,25 @@ +package com.example.aiapp + +/** + * What a machine's Claude Code sessions are having done to them, live. + * + * The import screen starts and then leaves work behind: the server runs it, so the phone that asked + * is free to go elsewhere and the answer arrives here rather than as a reply. What the screen shows + * on arrival comes from the listing, which carries the same state for whoever was not connected + * when it changed; this is only what keeps a screen somebody is watching current. + * + * The connection and its framing belong to [Sse]. Closing is the caller's cancellation path, and + * the caller owns reconnecting -- there is no cursor to resume from, because anything missed is in + * the next listing. + */ +class ImportableStream(settings: ServerSettings, private val setup: String) { + private val stream = Sse(settings) + + fun close() = stream.close() + + fun run(onOpen: () -> Unit, onChange: (ImportableChange) -> Unit) { + stream.run("/setups/$setup/importable/events", onOpen) { _, data -> + if (data.isNotEmpty()) parseImportableChange(data)?.let(onChange) + } + } +} diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt index e4cc8fa..ce673f8 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -81,8 +81,6 @@ import kotlinx.coroutines.flow.first import kotlinx.coroutines.launch import kotlinx.coroutines.withContext -private const val RECONNECT_DELAY_MS = 1500L - /** * How big the "still loading this conversation" spinner is. * diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Sse.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Sse.kt new file mode 100644 index 0000000..1e1897f --- /dev/null +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Sse.kt @@ -0,0 +1,97 @@ +package com.example.aiapp + +import java.io.IOException +import java.net.HttpURLConnection +import java.net.URL + +/** + * How long to wait before opening a dropped stream again. + * + * Shared by every screen that follows one, so a reconnect is not paced differently depending on + * which stream dropped. Short enough that a tunnel coming back is not noticed, long enough that a + * server which is genuinely down is not being asked several times a second. + */ +const val RECONNECT_DELAY_MS = 1500L + +/** + * One server-sent-events connection, framed. + * + * The framing is the part worth having once: `data:` and `event:` lines accumulate until a blank + * line ends the frame, comments (keep-alives) start with `:`, and a frame is either named with no + * payload or a payload with no name. Two screens follow two different streams — a session's + * transcript and what a machine's import list is doing — and neither should be re-deriving that. + * + * Blocking: [run] occupies its thread until the stream ends. [close], from any thread, is the + * cancellation path — it disconnects the socket, which unblocks the read, and [run] then returns + * rather than throwing, so a deliberate close is not reported as a connection error. Reconnecting + * belongs to the caller, which is the only one that knows where to resume from. + */ +class Sse(private val settings: ServerSettings) { + @Volatile private var connection: HttpURLConnection? = null + @Volatile private var closed = false + + fun close() { + closed = true + connection?.disconnect() + } + + /** + * Follows the stream at [path], handing each frame to [onFrame] as its name (null for an + * ordinary data frame) and its payload. The path is given here rather than at construction + * because a caller that reconnects usually resumes from somewhere new -- a cursor it has + * advanced past -- and that lives in the query string. + * + * [onOpen] fires once the server has accepted the connection. That is the measured moment the + * stream is live, and the only honest thing to clear a previous failure on: clearing on the + * first *event* instead left an idle stream displaying a connection error it had already + * recovered from, indefinitely. + */ + fun run(path: String, onOpen: () -> Unit, onFrame: (name: String?, data: String) -> Unit) { + val connection = URL("${settings.baseUrl}$path").openConnection() as HttpURLConnection + this.connection = connection + try { + connection.applyPinnedTls() + connection.connectTimeout = CONNECT_TIMEOUT_MS + // No read timeout: between events there is nothing to read for as long as the thing + // being followed is idle; the server's keep-alives and a dead socket erroring out are + // the liveness story. + connection.readTimeout = 0 + connection.setRequestProperty("Authorization", "Bearer ${settings.token}") + connection.setRequestProperty("Accept", "text/event-stream") + if (connection.responseCode != 200) { + val detail = connection.errorStream?.bufferedReader()?.readText()?.trim() + throw ApiException(detail ?: "HTTP ${connection.responseCode} for the event stream") + } + + onOpen() + val reader = connection.inputStream.bufferedReader() + val data = StringBuilder() + var name: String? = null + while (true) { + val line = reader.readLine() ?: break + when { + line.isEmpty() -> { + if (name != null || data.isNotEmpty()) onFrame(name, data.toString()) + data.clear() + name = null + } + line.startsWith("data:") -> data.append(line.removePrefix("data:").trim()) + line.startsWith("event:") -> name = line.removePrefix("event:").trim() + else -> {} // id:, comments -- nothing to do + } + } + } catch (e: ApiException) { + throw e + } catch (e: IOException) { + if (!closed) { + throw ApiException( + "Can't reach the server -- retrying. (${e.message ?: e::class.simpleName})", + e, + ) + } + } finally { + connection.disconnect() + this.connection = null + } + } +} diff --git a/app/ui-sandbox.sh b/app/ui-sandbox.sh index 85ae287..e229976 100755 --- a/app/ui-sandbox.sh +++ b/app/ui-sandbox.sh @@ -22,9 +22,10 @@ # ./ui-sandbox.sh start it, print the enrolment command # ./ui-sandbox.sh stop stop it # -# Environment: AI_SANDBOX_ROOT, AI_SANDBOX_TOKEN, AI_SANDBOX_PORT, and -# AI_SANDBOX_DELAY -- the last being the server's own `--delay`, which is -# what makes a spinner visible at all. On loopback every request is back in +# Environment: AI_SANDBOX_ROOT, AI_SANDBOX_TOKEN, AI_SANDBOX_PORT, +# AI_SANDBOX_DELAY -- the server's own `--delay`, which is what makes a +# spinner visible at all -- and AI_SANDBOX_SPAWN_DELAY, which holds an +# import open for that many seconds. On loopback every request is back in # under a millisecond, so a busy state that is correct is still a busy state # nobody can see. set -eu @@ -33,6 +34,8 @@ ROOT=${AI_SANDBOX_ROOT:-${XDG_RUNTIME_DIR:-/tmp}/ai-app-sandbox} TOKEN=${AI_SANDBOX_TOKEN:-sandbox} PORT=${AI_SANDBOX_PORT:-8443} DELAY=${AI_SANDBOX_DELAY:-1200} +SPAWN_DELAY=${AI_SANDBOX_SPAWN_DELAY:-0} +BIG_MB=${AI_SANDBOX_BIG_MB:-40} CERTS=${AI_SANDBOX_CERTS:-${XDG_CONFIG_HOME:-$HOME/.config}/ai-app/certs} SCRIPT_DIR=$(cd "$(dirname "$0")" && pwd) @@ -96,12 +99,37 @@ done # records a real pid, writes nothing, and dies on a signal. A real # `claude --resume` against an invented session id would either fail in a # way that tests nothing or start a turn on somebody's account. -cat >"$ROOT/fake-claude" <<'FAKE' +cat >"$ROOT/fake-claude" < /dev/null FAKE chmod +x "$ROOT/fake-claude" +# One big one, because size is what makes importing take any time at all. +# A spawn replays the whole file into this app's transcript, so against the +# four-line sessions above it is over in milliseconds and every state on the +# way is unobservable -- which is how a row that should have been marked +# "importing" went unnoticed for not being marked at all. AI_SANDBOX_BIG_MB +# sets how large. +big=$PROJECTS/0000000b-5eed-4a11-9c0d-00000000b000.jsonl +awk -v mb="$BIG_MB" 'BEGIN { + target = mb * 1000000 + line = "{\"type\":\"user\",\"cwd\":\"/home/bob/repos/sandbox/big\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"text\",\"text\":\"a long sandbox turn, number %d, with enough text on it that the file reaches a realistic size rather than a token one\"}]}}" + written = 0 + for (i = 1; written < target; i++) { + out = sprintf(line, i) + print out + written += length(out) + 1 + } + print "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"usage\":{\"input_tokens\":180000,\"output_tokens\":900}}}" +}' > "$big" + hash=$(printf '%s' "$TOKEN" | sha256sum | cut -d' ' -f1) cat >"$ROOT/config.ron" <) -> Router { "/setups/{id}/importable/{session}", delete(delete_importable), ) + .route( + "/setups/{id}/importable/{session}/import", + post(start_import), + ) + // Static segment, so this wins over `{session}` above rather than + // being read as a session called "events". + .route("/setups/{id}/importable/events", get(importable_events)) .route( "/setups/{id}", get(read_setup).put(update_setup).delete(delete_setup), @@ -450,7 +458,7 @@ struct SpawnRequest { async fn list_importable( State(manager): State>, UrlPath(id): UrlPath, -) -> Result>, ApiError> { +) -> Result>, ApiError> { let setup = setup_by_id(&manager, &id)?; let transport = crate::session::transport::Transport::for_setup(&setup); let mut found = crate::session::import::list(&transport) @@ -464,8 +472,59 @@ async fn list_importable( // Joined here because the importer knows about files and the manager // knows about sessions, and putting the two together is the route's // job rather than either one's. - found.retain(|candidate| manager.session_driving(&candidate.id).is_none()); - Ok(axum::Json(found)) + // + // Except while this server is in the middle of importing it. A spawn + // creates the session partway through, so the row would vanish the + // instant the work started and reappear as a session only once it + // finished -- and in between, the screen that asked for it would be + // showing nothing at all where the thing it is waiting for used to be. + // A row with an operation on it stays until the operation settles. + found.retain(|candidate| { + manager.pending().running(&id, &candidate.id).is_some() + || manager.session_driving(&candidate.id).is_none() + }); + + // What the server is doing to each of them, joined on here because a + // phone that was asleep, out of range, or freshly opened never heard + // the events -- see `pending`. An operation is *not* filtered out + // above: a row being imported has to stay visible, marked, or the list + // would say the work never started. + let present: Vec = found.iter().map(|row| row.id.clone()).collect(); + manager.pending().prune(&id, &present); + let rows: Vec = found + .into_iter() + .map(|importable| ImportableRow { + pending: manager + .pending() + .running(&id, &importable.id) + .map(|operation| operation.label()), + error: manager.pending().failure(&id, &importable.id), + importable, + }) + .collect(); + Ok(axum::Json(rows)) +} + +/// A row of the import list: what the machine has, plus what this server is +/// doing to it. +/// +/// Flattened, so the two halves arrive as one object -- the phone is +/// drawing one row and has no use for the seam between "what the machine +/// said" and "what we are doing about it". +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ImportableRow { + #[serde(flatten)] + importable: crate::session::import::Importable, + /// The word the row shows while something is running: "importing" or + /// "deleting". Absent when nothing is. + #[serde(skip_serializing_if = "Option::is_none")] + pending: Option<&'static str>, + /// How the last attempt on this row failed, if it did. Kept until + /// something replaces it, because the phone that needs to see it may + /// not have been connected when it happened. + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, } /// Removes a Claude Code session from a machine. @@ -480,30 +539,143 @@ async fn delete_importable( ) -> Result { let setup = setup_by_id(&manager, &id)?; let transport = crate::session::transport::Transport::for_setup(&setup); - crate::session::import::delete(&transport, &session) - .await - .map_err(bad_request)?; - tracing::info!("deleted Claude Code session {session} from setup {id}"); - Ok(StatusCode::NO_CONTENT) + let target = session.clone(); + in_background(&manager, id, session, Operation::Deleting, async move { + crate::session::import::delete(&transport, &target).await + }); + Ok(StatusCode::ACCEPTED) +} + +/// Continues a Claude Code session, in the background. +/// +/// Separate from `POST /sessions` because the two are asked different +/// questions. That one means "start this and take me to it", so it waits +/// and answers with the session. This one is the import screen's batch: +/// several at once, nobody waiting on any particular one, and the answer +/// arrives as a row changing rather than as a reply -- which is the whole +/// point, since the screen it was started from may well be gone by then. +async fn start_import( + State(manager): State>, + UrlPath((id, session)): UrlPath<(String, String)>, + axum::Json(body): axum::Json, +) -> Result { + // Checked before accepting, so an unknown machine is still an error the + // caller sees rather than a failure it has to go and read off a row. + setup_by_id(&manager, &id)?; + let request = SpawnRequest { + setup: id.clone(), + provider: body.provider, + // Nothing to say: `spawn` titles an import from the session it + // continues, and the cwd comes from the same place. + title: None, + model: body.model, + cwd: None, + permission_mode: body.permission_mode, + params: std::collections::BTreeMap::new(), + import: Some(session.clone()), + }; + let inner = Arc::clone(&manager); + in_background(&manager, id, session, Operation::Importing, async move { + spawn(&inner, request) + .await + .map(|_| ()) + .map_err(|err| anyhow::anyhow!("{err}")) + }); + Ok(StatusCode::ACCEPTED) +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +#[serde(deny_unknown_fields)] +struct ImportRequest { + provider: String, + #[serde(default)] + model: Option, + #[serde(default)] + permission_mode: Option, +} + +/// Runs `work` on the server, marked as in flight for as long as it takes. +/// +/// Spawned rather than awaited, which is the whole difference: the phone +/// asked for it, but the phone leaving must not cancel it. What replaces +/// the reply is the pending registry -- the row says what is happening to +/// it, whoever is looking and whenever they look. +fn in_background( + manager: &Arc, + setup: String, + session: String, + operation: Operation, + work: F, +) where + F: std::future::Future> + Send + 'static, +{ + let running = manager.pending().begin(&setup, &session, operation); + tokio::spawn(async move { + match work.await { + Ok(()) => { + tracing::info!("{} {session} on {setup}: done", operation.label()); + running.succeeded(); + } + Err(err) => { + tracing::warn!("{} {session} on {setup} failed: {err:#}", operation.label()); + // The server's own words, the way every other failure in + // this app reaches a person. + running.failed(format!("{err:#}")); + } + } + }); +} + +/// Every change to what is in flight against one machine. +/// +/// Scoped to the setup the screen is showing, the same way a session's +/// events are scoped to that session -- a phone watching one machine's +/// import list has no use for another's. +async fn importable_events( + State(manager): State>, + UrlPath(id): UrlPath, +) -> Sse>> { + let live = manager.pending().subscribe(); + let stream = BroadcastStream::new(live).filter_map(move |item| { + // A lagged subscriber has missed changes it cannot get back here, + // and that is what the listing is for: the screen refetches on + // arrival and carries the truth whatever this stream missed. + let change = item.ok()?; + if change.setup() != id { + return None; + } + Some(Ok(SseEvent::default().json_data(&change).ok()?)) + }); + Sse::new(stream).keep_alive(KeepAlive::default()) } async fn spawn_session( State(manager): State>, axum::Json(body): axum::Json, ) -> Result, ApiError> { + spawn(&manager, body).await.map(axum::Json) +} + +/// Starts a session, continuing a Claude Code one where `body.import` names +/// it. +/// +/// A function rather than only a handler because the import screen's batch +/// runs this from a background task -- see [`start_import`]. Spawning has to +/// mean exactly the same thing either way: the same refusal when something +/// else already has the conversation open, the same title, the same working +/// directory. +async fn spawn(manager: &Arc, body: SpawnRequest) -> Result { // Resolved before the spawn because both halves of it are the // machine's answer, not the phone's: which file that id names, and // what is in it. let seed = match &body.import { Some(want) => { - let setup = setup_by_id(&manager, &body.setup)?; + let setup = setup_by_id(manager, &body.setup)?; let transport = crate::session::transport::Transport::for_setup(&setup); - let found = crate::session::import::list(&transport) + let chosen = crate::session::import::find(&transport, want) .await - .map_err(bad_request)?; - let chosen = found - .into_iter() - .find(|candidate| &candidate.id == want) + .map_err(bad_request)? .ok_or_else(|| { ApiError::NotFound(format!( "setup \"{}\" has no Claude Code session {want} to import", @@ -615,7 +787,7 @@ async fn spawn_session( info.id, info.title ); - Ok(axum::Json(info)) + Ok(info) } #[derive(Deserialize)] diff --git a/server/src/session/import.rs b/server/src/session/import.rs index 7513db8..18ffb6c 100644 --- a/server/src/session/import.rs +++ b/server/src/session/import.rs @@ -166,7 +166,53 @@ pub async fn list(transport: &Transport) -> Result> { // its text in a list, so that reading lost twenty rows rather than // two. Excluding `tool_use_id` keeps both shapes of a real message // and drops the one that is not. - let script = r#" + let script = listing_script(r#""$HOME"/.claude/projects/*/*.jsonl"#); + let launch = Launch::new("sh", vec!["-c".to_string(), script], None); + parse_listing(&transport.capture(&launch).await?) +} + +/// The same listing, for one session named by id. +/// +/// Importing needs everything a row holds -- the path to follow, how many +/// lines have already been written, what it is called, where it was working +/// and whether something else has it open -- and used to get them by +/// listing *every* session and searching the result. That is a full read of +/// every transcript on the machine, seconds of it, to answer a question +/// about one file; a batch of imports paid it once each. Same script, same +/// parsing, one glob narrower. +pub async fn find(transport: &Transport, id: &str) -> Result> { + if !is_session_id(id) { + return Ok(None); + } + let script = listing_script(r#""$HOME"/.claude/projects/*/"$1".jsonl"#); + let launch = Launch::new( + "sh", + vec!["-c".to_string(), script, "sh".to_string(), id.to_string()], + None, + ); + Ok(parse_listing(&transport.capture(&launch).await?)? + .into_iter() + .find(|candidate| candidate.id == id)) +} + +/// What the machine is asked, over whichever set of files `glob` names. +/// +/// One script with the glob substituted rather than two that drift: the +/// per-file half decides what a row *is*, and a row has to mean the same +/// thing whether it arrived from a listing or from a lookup. The glob is +/// this module's own text; the only thing that ever crosses from outside is +/// the id, which stays an argument (`$1`) and is checked by +/// [`is_session_id`] first. +fn listing_script(glob: &str) -> String { + // `replace` rather than `format!`: this is shell, so it is full of + // braces -- `${s##*/}`, an awk program, the `{[^}]*` that finds a usage + // record -- and every one of them would have to be doubled to survive a + // format string. Doubling braces inside a script is exactly the kind of + // edit that looks right and changes what the shell runs. + SCRIPT.replace("{glob}", glob) +} + +const SCRIPT: &str = r#" if [ -d "$HOME/.claude/sessions" ]; then printf 'LIVEKNOWN\n' for s in "$HOME"/.claude/sessions/*.json; do @@ -180,7 +226,7 @@ if [ -d "$HOME/.claude/sessions" ]; then [ -n "$sid" ] && printf 'LIVE\t%s\n' "$sid" done fi -for f in "$HOME"/.claude/projects/*/*.jsonl; do +for f in {glob}; do [ -f "$f" ] || continue printf '%s\t%s\t%s\t%s\t%s\t' "$(stat -c %Y "$f" 2>/dev/null || echo 0)" \ "$(wc -l < "$f")" "$(stat -c %s "$f" 2>/dev/null || echo 0)" \ @@ -190,9 +236,9 @@ for f in "$HOME"/.claude/projects/*/*.jsonl; do printf '\n' done "#; - let launch = Launch::new("sh", vec!["-c".to_string(), script.to_string()], None); - let found = transport.capture(&launch).await?; +/// Rows out of what [`listing_script`] printed, with `in_use` filled in. +fn parse_listing(found: &str) -> Result> { let mut live = std::collections::HashSet::new(); let mut checkable = false; for line in found.lines() { diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 8445756..4fbac3b 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -14,6 +14,7 @@ pub mod driver; pub mod echo; pub mod import; pub mod llama; +pub mod pending; pub mod process; pub mod transcript; pub mod transport; @@ -530,6 +531,10 @@ pub struct SessionManager { /// Held here rather than per session for the reason /// [`SessionManager::subscribe_notifications`] gives. notifications: broadcast::Sender, + /// Imports and deletes running against a machine's Claude Code + /// sessions. Beside the notification channel above because it is the + /// same kind of thing: state the phone reads but does not own. + pending: Arc, /// What to mark sessions spawned here as -- see /// [`SessionManager::marking_new_sessions_throwaway`] and /// [`SessionConfig::throwaway`]. @@ -585,6 +590,7 @@ impl SessionManager { data_dir, models_dir, notifications, + pending: Arc::new(pending::Registry::default()), spawn_throwaway: false, inner: RwLock::new(Inner { config, live }), }; @@ -979,6 +985,13 @@ impl SessionManager { self.notifications.subscribe() } + /// Imports and deletes running against importable sessions -- see + /// [`pending::Registry`], which is also where the reason it lives on + /// the server rather than in the phone is written down. + pub fn pending(&self) -> &Arc { + &self.pending + } + pub fn session(&self, id: &str) -> Option> { self.inner.read().unwrap().live.get(id).cloned() } diff --git a/server/src/session/pending.rs b/server/src/session/pending.rs new file mode 100644 index 0000000..86a3487 --- /dev/null +++ b/server/src/session/pending.rs @@ -0,0 +1,282 @@ +//! What is being done to a machine's Claude Code sessions right now. +//! +//! Importing and deleting used to be whatever the phone was in the middle +//! of: the request was the work, so leaving the screen cancelled it and +//! coming back showed no sign it had ever started. Sessions half-imported +//! that way are the expensive kind of missing -- the row is back in the +//! list looking untouched, and taking it again is the second `--resume` the +//! whole import path exists to prevent. +//! +//! So the work runs here, on the server, and this is the record of it. The +//! phone reads that record two ways, and needs both: every row of `GET +//! /setups/{id}/importable` carries what is happening to it, which is what +//! a phone that was asleep, out of range, or freshly opened has to go on; +//! and [`Registry::subscribe`] is the live stream, which is what makes a +//! screen somebody is looking at change by itself. Neither is sufficient +//! alone -- a broadcast has no memory, and a listing is only true when it +//! was fetched. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use serde::Serialize; +use tokio::sync::broadcast; + +/// What is being done to an importable session. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub enum Operation { + Importing, + Deleting, +} + +impl Operation { + /// The word a row shows while this runs. Fixed here rather than in the + /// app so the two ends cannot disagree about what a state is called. + pub fn label(self) -> &'static str { + match self { + Self::Importing => "importing", + Self::Deleting => "deleting", + } + } +} + +/// One change to what is in flight, as it goes out on the stream. +/// +/// The three states are every way an operation ends, including the two that +/// are easy to leave out: it can still be running, it can have finished, +/// and it can have failed. There is deliberately no "unknown" -- this is +/// the server's own work, so not knowing would be a bug rather than a +/// state. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase", tag = "state")] +pub enum Change { + Started { + setup: String, + session: String, + operation: Operation, + }, + Finished { + setup: String, + session: String, + }, + Failed { + setup: String, + session: String, + message: String, + }, +} + +impl Change { + /// Which machine this is about, so a stream scoped to one can drop the + /// rest. Every variant carries it; matching here rather than at the + /// filter keeps that fact in one place. + pub fn setup(&self) -> &str { + match self { + Self::Started { setup, .. } + | Self::Finished { setup, .. } + | Self::Failed { setup, .. } => setup, + } + } +} + +/// Everything in flight, and the last failure against each session. +#[derive(Debug)] +pub struct Registry { + running: Mutex>, + /// Kept after the operation ends, because a phone that was not looking + /// when it failed has no other way to find out. Replaced when the next + /// operation on that session starts, and dropped by [`Registry::prune`] + /// when the session is no longer on the machine -- an error about a + /// transcript that is gone has nothing left to be about. + failures: Mutex>, + changes: broadcast::Sender, +} + +impl Default for Registry { + fn default() -> Self { + Self { + running: Mutex::new(HashMap::new()), + failures: Mutex::new(HashMap::new()), + // Enough that a phone watching one screen cannot lag behind a + // batch of any size somebody would start by hand. + changes: broadcast::channel(256).0, + } + } +} + +impl Registry { + /// Marks an operation as running and announces it. + /// + /// The returned guard is how it stops being marked: settle it with + /// [`InFlight::succeeded`] or [`InFlight::failed`], or drop it and it + /// reports a failure. Dropping without settling means the task was + /// cancelled or panicked, and a row stuck on "importing" for ever is a + /// worse answer than one that says it did not finish. + pub fn begin(self: &Arc, setup: &str, session: &str, operation: Operation) -> InFlight { + let key = (setup.to_string(), session.to_string()); + self.running.lock().unwrap().insert(key.clone(), operation); + self.failures.lock().unwrap().remove(&key); + let _ = self.changes.send(Change::Started { + setup: key.0.clone(), + session: key.1.clone(), + operation, + }); + InFlight { + registry: Arc::clone(self), + key, + settled: false, + } + } + + /// What is happening to this session, if anything is. + pub fn running(&self, setup: &str, session: &str) -> Option { + let key = (setup.to_string(), session.to_string()); + self.running.lock().unwrap().get(&key).copied() + } + + /// How the last operation on this session failed, if it did. + pub fn failure(&self, setup: &str, session: &str) -> Option { + let key = (setup.to_string(), session.to_string()); + self.failures.lock().unwrap().get(&key).cloned() + } + + /// Forgets failures against sessions the machine no longer has. + /// + /// Called from the listing, which is the only place that knows what is + /// still there. A deleted session's failure would otherwise outlive + /// everything it referred to. + pub fn prune(&self, setup: &str, present: &[String]) { + self.failures + .lock() + .unwrap() + .retain(|(kept_setup, session), _| { + kept_setup != setup || present.iter().any(|id| id == session) + }); + } + + /// Every change as it happens. See the module note on why this is not + /// the only way the phone finds out. + pub fn subscribe(&self) -> broadcast::Receiver { + self.changes.subscribe() + } +} + +/// An operation that is running, and its way back out of the registry. +pub struct InFlight { + registry: Arc, + key: (String, String), + settled: bool, +} + +impl InFlight { + pub fn succeeded(mut self) { + self.settle(None); + } + + pub fn failed(mut self, message: String) { + self.settle(Some(message)); + } + + fn settle(&mut self, failure: Option) { + if self.settled { + return; + } + self.settled = true; + self.registry.running.lock().unwrap().remove(&self.key); + let (setup, session) = (self.key.0.clone(), self.key.1.clone()); + let change = match failure { + Some(message) => { + self.registry + .failures + .lock() + .unwrap() + .insert(self.key.clone(), message.clone()); + Change::Failed { + setup, + session, + message, + } + } + None => Change::Finished { setup, session }, + }; + let _ = self.registry.changes.send(change); + } +} + +impl Drop for InFlight { + fn drop(&mut self) { + self.settle(Some("the server stopped before it finished".to_string())); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn an_operation_is_visible_while_it_runs_and_gone_after() { + let registry = Arc::new(Registry::default()); + let mut changes = registry.subscribe(); + + let running = registry.begin("local", "abc", Operation::Importing); + assert_eq!(registry.running("local", "abc"), Some(Operation::Importing)); + assert!(matches!(changes.try_recv(), Ok(Change::Started { .. }))); + + running.succeeded(); + assert_eq!(registry.running("local", "abc"), None); + assert_eq!(registry.failure("local", "abc"), None); + assert!(matches!(changes.try_recv(), Ok(Change::Finished { .. }))); + } + + /// A failure outlives the operation, because the phone that needs it may + /// not have been listening when it happened. + #[test] + fn a_failure_is_kept_until_something_replaces_or_prunes_it() { + let registry = Arc::new(Registry::default()); + + registry + .begin("local", "abc", Operation::Deleting) + .failed("no such session".to_string()); + assert_eq!(registry.running("local", "abc"), None); + assert_eq!( + registry.failure("local", "abc").as_deref(), + Some("no such session") + ); + + // Still on the machine, so the failure is still about something. + registry.prune("local", &["abc".to_string()]); + assert!(registry.failure("local", "abc").is_some()); + + // Another machine's listing says nothing about this one's. + registry.prune("other", &[]); + assert!(registry.failure("local", "abc").is_some()); + + registry.prune("local", &[]); + assert!(registry.failure("local", "abc").is_none()); + } + + /// Trying again clears the last failure, so a row cannot show an error + /// from before the attempt somebody is currently watching. + #[test] + fn starting_again_clears_the_previous_failure() { + let registry = Arc::new(Registry::default()); + registry + .begin("local", "abc", Operation::Importing) + .failed("first go".to_string()); + + let second = registry.begin("local", "abc", Operation::Importing); + assert_eq!(registry.failure("local", "abc"), None); + second.succeeded(); + } + + /// A task that is cancelled or panics must not leave a row saying + /// something is still happening to it. + #[test] + fn dropping_an_unsettled_operation_reports_a_failure() { + let registry = Arc::new(Registry::default()); + drop(registry.begin("local", "abc", Operation::Deleting)); + assert_eq!(registry.running("local", "abc"), None); + assert!(registry.failure("local", "abc").is_some()); + } +}