From 3c159fa1e1aa0813b7e013ab2627a71a642a3b05 Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Mon, 31 Aug 2026 21:05:33 -0400 Subject: [PATCH] Run imports and deletes on the server, and say so on an event Leaving the import screen used to cancel the batch it had started: the request was the work, so the coroutine that owned it died with the screen and coming back showed no sign anything had happened. A half-imported session is the expensive kind of missing -- the row is back looking untouched, and taking it again is the second `--resume` the import path exists to prevent. So the work runs on the server now. Delete and a new per-session import both answer 202 and spawn the work, and `session::pending` is the record of it: what is running, and how the last attempt failed. The phone reads that two ways and needs both. Every row of the listing carries `pending` and `error`, which is what a phone that was asleep, out of range or freshly opened has to go on; `GET /setups/{id}/importable/events` streams the changes, which is what makes a screen somebody is watching change by itself. Neither alone is enough, and that is not theoretical. A broadcast has no memory, so an operation that started and finished while the stream was still connecting was one nothing would ever be said about -- with responses held back far enough to make it visible, one row of a pair of deletes cleared and the other sat on "waiting" for good. The screen now asks again after a handover when anything still looks outstanding, and takes its row states from that answer rather than from what it remembers. The single tap still waits, because "take me to it" needs the session that was made and 202 does not carry one. Both paths go through the same `spawn` so they cannot drift about what importing means. Resolving one importable session no longer lists every one of them: `import::find` is the same script with one glob narrower, which takes the import seed off the 3.7-second full scan that `delete` came off earlier. The SSE connection and its framing are now `Sse`, shared with the session transcript stream rather than written a second time. --- AGENTS.md | 31 +- .../src/main/kotlin/com/example/aiapp/Api.kt | 81 +++++ .../kotlin/com/example/aiapp/EventStream.kt | 85 +----- .../kotlin/com/example/aiapp/ImportScreen.kt | 256 +++++++++++----- .../com/example/aiapp/ImportableStream.kt | 25 ++ .../kotlin/com/example/aiapp/SessionScreen.kt | 2 - .../src/main/kotlin/com/example/aiapp/Sse.kt | 97 ++++++ app/ui-sandbox.sh | 38 ++- server/src/routes.rs | 204 ++++++++++++- server/src/session/import.rs | 54 +++- server/src/session/mod.rs | 13 + server/src/session/pending.rs | 282 ++++++++++++++++++ 12 files changed, 992 insertions(+), 176 deletions(-) create mode 100644 app/androidApp/src/main/kotlin/com/example/aiapp/ImportableStream.kt create mode 100644 app/androidApp/src/main/kotlin/com/example/aiapp/Sse.kt create mode 100644 server/src/session/pending.rs 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()); + } +}