diff --git a/AGENTS.md b/AGENTS.md index f94676b..9279365 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -205,6 +205,25 @@ day to day: in `process.json`; removing either by hand while the session is live loses output or replays it. +## Auto-resume + +**A session switched to it sends itself a message once the account's usage +limit lifts** — off by default, per session, in the session settings dialog. +PLAN.md's "Auto-resume" is the design; day to day: + +- **The schedule is a plan to ask.** `resume.rs` wakes at the scheduled time, + asks `GET /usage`'s meter for that machine and provider, and only sends when + it answers `ok` with nothing at 100%. Anything else — still spent, logged + out, unreachable — is a longer wait, and a still-spent window reschedules to + the reset time the *meter* now gives. +- **Test it with echo, never with a real account.** `/limit [minutes]` reports + the same `limitReached` event a real driver does, and `/usage 100 5` sets + what the meter answers. They are deliberately separate: the two disagreeing + is the case the design exists for. `/usage 20` is the limit lifting. +- The wait is on the session in `config.ron` (`resume`), so it survives a + backend restart. A day after the limit was hit it gives up and says so in + the transcript. + ## Shared appearance - **A row something is happening to is dimmed, drained of colour, and says diff --git a/PLAN.md b/PLAN.md index 2beee6d..3f1a0a0 100644 --- a/PLAN.md +++ b/PLAN.md @@ -596,6 +596,54 @@ always running. So absent means **not running**, and only a timestamp that arrives and cannot be parsed is unknown. `WindowEnd` in `ResetCountdown.kt` is the one rule both readers go through. +### Auto-resume (2026-09-05) + +**A session may pick itself back up when the account's usage limit lifts.** +Off unless somebody switched that session to it, because it spends quota the +moment quota exists and does so with nobody looking — that is not a thing a +default may decide. It sends one message, `continue` unless another was +typed, and then it is done; there is no retry loop around the conversation +itself. + +**Running out of quota is a state, not an error.** `Event::LimitReached` +carries the dialect's reset time where it gave one, and recognising it +belongs to the driver — the Claude CLI ends the turn with `is_error` and +`Claude AI usage limit reached|1788546972`, and nothing above the driver +matches on a string. The transcript draws it as a divider, like a clear or a +compaction: what a reader scrolling back wants from it is why the +conversation stops at that line. + +**The schedule is a plan to ask, never a plan to send.** Every reset time +available here is untrustworthy in the direction that matters: the dialect's +is written when the turn fails, and the endpoint's moves when the window +does. So the wait ends in a question to `usage.rs`, and only `ok` with no +window at 100% sends anything. A window still spent reschedules to *its own* +reset time — which is what makes a limit that lifts later than promised wait +longer, and one that lifts sooner resume sooner. A meter that cannot be +asked at all is a longer wait too, never a send: "we could not find out" +must not be able to produce the same action as "there is room". + +Bounded, because something has to be: a day after the limit was hit the wait +stops and says so in the session's own transcript. A machine that can never +be asked would otherwise be retried for ever with nothing on screen saying +so. + +The schedule is persisted on the session (`resume: Some(ScheduledResume)`), +not held in memory: a five-hour window routinely outlasts a backend restart, +and a wait forgotten across one is a session that silently never comes back. +`resume.rs` is the top layer — it holds the manager and the monitor and +neither holds it — which is what lets the decision be a pure function of a +snapshot and a clock. The pump reports limits downward on a broadcast, for +the reason `Shared` exists: the pump runs underneath the manager. + +**Exercised with echo, never with a real account.** `/limit [minutes]` in an +echo session reports the same event a real driver does, and `/usage` sets +what the meter answers — deliberately two commands, because the two +disagreeing is the state the whole design is about. The loop was driven end +to end that way on 2026-09-05: the wait moved from the dialect's two minutes +to the meter's seven when the meter changed its mind, and the message went +out on the first check after the meter came back under the limit. + ### HTTP surface **`routes.rs`'s module doc comment is the table.** REST for actions, one SSE 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 c7fb46f..d64df02 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -167,6 +167,24 @@ data class SessionSummary( * itself from a default is one you can turn off while believing you are reading it. */ val notify: Boolean, + /** + * Whether this session sends itself a message once its account's usage limit lifts, and what + * that message says. + * + * The message is what the server would actually send, with its own default already filled in, + * so the field shows the words rather than an empty box standing for them. + */ + val autoResume: Boolean, + val autoResumeMessage: String, + /** + * When the server next intends to check whether the limit has lifted, in epoch seconds, or null + * when nothing is waiting. + * + * A time to *ask*, not a time to resume: the server checks the meter at that moment and waits + * again if the limit is still on. Worded that way wherever it is shown, because a promise this + * app cannot keep is worse than no time at all. + */ + val resumeAt: Double?, /** * The directory the session works in, or null where it was never given one. * @@ -221,6 +239,12 @@ private fun parseSession(session: JSONObject) = takesEffort = session.optBoolean("takesEffort", false), imported = session.optBoolean("imported", false), notify = session.optBoolean("notify", true), + autoResume = session.optBoolean("autoResume", false), + // The server sends its own default rather than nothing, so an empty answer means an older + // server -- and this app's word for it is the same word. + autoResumeMessage = + session.optString("autoResumeMessage").ifEmpty { DEFAULT_RESUME_MESSAGE }, + resumeAt = if (session.has("resumeAt")) session.getDouble("resumeAt") else null, cwd = session.optString("cwd").ifEmpty { null }, contextTokens = if (session.has("contextTokens")) session.getLong("contextTokens") else null, @@ -1072,6 +1096,36 @@ fun setSessionPermissionMode(settings: ServerSettings, sessionId: String, mode: } /** Turns this session's notifications on or off. Stored on the backend -- see `SessionConfig`. */ +/** + * What an auto-resume says when nothing else was typed. Mirrors the server's own default, so a + * cleared field shows the word that would actually be sent instead of going blank. + */ +const val DEFAULT_RESUME_MESSAGE = "continue" + +/** + * Turns auto-resume on or off and sets what it would say, in one request because they are one + * decision -- see the server's `/sessions/{id}/auto-resume`. + */ +fun setSessionAutoResume( + settings: ServerSettings, + sessionId: String, + autoResume: Boolean, + message: String?, +) { + requestFromServer( + settings, + "/sessions/$sessionId/auto-resume", + method = "POST", + jsonBody = + JSONObject() + .put("autoResume", autoResume) + // Empty means the server's default rather than a session poked with nothing to + // read, which is the same rule the server applies to the field. + .put("message", message?.trim()?.ifEmpty { null } ?: JSONObject.NULL) + .toString(), + ) {} +} + fun setSessionNotify(settings: ServerSettings, sessionId: String, notify: Boolean) { requestFromServer( settings, diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Dividers.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Dividers.kt index 01a13ab..57600e1 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Dividers.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Dividers.kt @@ -12,6 +12,10 @@ import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.graphics.Color import androidx.compose.ui.unit.dp +import java.time.Instant +import java.time.ZoneId +import java.time.format.DateTimeFormatter +import java.time.format.FormatStyle /** * A line across the transcript saying what left the session's context. @@ -47,3 +51,40 @@ fun TranscriptDivider(text: String, color: Color, modifier: Modifier = Modifier) fun ClearedRow(modifier: Modifier = Modifier) { TranscriptDivider("Context cleared", clearedColor, modifier) } + +/** + * The mark running out of quota leaves. + * + * The same red the usage bar takes when a window is spent, because it is the same fact in a second + * place: colour by consequence, so "there is nothing left to spend" is learned once. + * + * A time rather than a countdown. The row is folded once and never re-measured, so a span would go + * stale on screen the moment it was drawn; and this is when the *account* said it would reset, + * which is not a promise about when the session picks back up. A limit the session was told no + * reset time for says nothing about one -- that state has its own words rather than a plausible + * number. + */ +@Composable +fun LimitRow(item: TranscriptItem.LimitNote, modifier: Modifier = Modifier) { + TranscriptDivider(limitSummary(item.resetsAt, ZoneId.systemDefault()), overLimitColor, modifier) +} + +/** + * What the row says. Split out so the wording is testable without a screen, since the two states it + * has to keep apart -- a reset time that arrived and one that never did -- are exactly the pair + * that reads the same when it goes wrong. + * + * [zone] is a parameter rather than read here so a test says the same thing wherever it runs. + */ +fun limitSummary(resetsAt: Double?, zone: ZoneId): String { + val at = resetsAt?.let { + try { + DateTimeFormatter.ofLocalizedTime(FormatStyle.SHORT) + .withZone(zone) + .format(Instant.ofEpochSecond(it.toLong())) + } catch (_: Exception) { + null + } + } + return if (at == null) "Usage limit reached" else "Usage limit reached • resets $at" +} diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt index f266f70..8580867 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt @@ -157,6 +157,18 @@ sealed class SessionEvent { */ data object Cleared : SessionEvent() + /** + * The session stopped because its account's usage limit was reached. + * + * Its own event rather than an [Error] carrying the CLI's sentence, because it is a state + * rather than something that went wrong -- and because the raw sentence is `Claude AI usage + * limit reached|1788546972`, which is not readable by the person it is shown to. + * + * [resetsAt] is epoch seconds and null where the session was told nothing. Only the server acts + * on it; what this draws it as is a time, not a countdown, because nothing here re-measures it. + */ + data class LimitReached(val resetsAt: Double?) : SessionEvent() + data class Error(val message: String) : SessionEvent() /** @@ -261,6 +273,10 @@ fun parseSeqEvent(json: String): SeqEvent { trigger = body.optString("trigger").ifEmpty { null }, ) "cleared" -> SessionEvent.Cleared + "limitReached" -> + SessionEvent.LimitReached( + if (body.has("resetsAt")) body.getDouble("resetsAt") else null + ) "error" -> SessionEvent.Error(body.getString("message")) else -> SessionEvent.Unknown(type) } 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 f31cd0e..64e031c 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -1545,6 +1545,7 @@ fun SessionScreen( is TranscriptItem.ClearedNote -> ClearedRow() is TranscriptItem.CompactedNote -> CompactedRow(item) + is TranscriptItem.LimitNote -> LimitRow(item) // Never reached: a peer message is flattened into // its own units. Here because a `when` over the // item kinds has to stay exhaustive. diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt index c756642..0a68e41 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsDialog.kt @@ -6,8 +6,10 @@ import androidx.compose.foundation.layout.Spacer import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.height import androidx.compose.foundation.layout.width +import androidx.compose.foundation.rememberScrollState import androidx.compose.foundation.text.KeyboardActions import androidx.compose.foundation.text.KeyboardOptions +import androidx.compose.foundation.verticalScroll import androidx.compose.material3.AlertDialog import androidx.compose.material3.CircularProgressIndicator import androidx.compose.material3.MaterialTheme @@ -26,6 +28,10 @@ import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.text.input.ImeAction import androidx.compose.ui.unit.dp +import java.time.Instant +import java.time.ZoneId +import java.time.format.DateTimeFormatter +import java.time.format.FormatStyle import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.launch import kotlinx.coroutines.withContext @@ -89,6 +95,16 @@ fun SessionSettingsDialog( // and a spinner sits beside it, which is what not knowing looks like. var notify by remember(sessionId) { mutableStateOf(null) } var notifyError by remember { mutableStateOf(null) } + // The same three-state shape the notification switch has, for the same reason: until the + // server has answered, the switch is disabled rather than showing a position nothing confirmed. + var autoResume by remember(sessionId) { mutableStateOf(null) } + var resumeMessage by remember(sessionId) { mutableStateOf(DEFAULT_RESUME_MESSAGE) } + // When the server next intends to ask whether the limit has lifted, or null when nothing is + // waiting. Read once with everything else: it moves on the server's schedule, not this + // screen's, and a figure that redrew itself here would be this app re-measuring what it was + // told. + var resumeAt by remember(sessionId) { mutableStateOf(null) } + var resumeError by remember { mutableStateOf(null) } // Where the session works. Null until the server has been asked, for the same reason the switch // above is. An empty answer is a session that was never given a directory, which is not the // same as one whose directory is unknown -- the field is only enabled once one of those is @@ -102,6 +118,9 @@ fun SessionSettingsDialog( try { val fresh = withContext(Dispatchers.IO) { fetchSession(settings, sessionId) } notify = fresh.notify + autoResume = fresh.autoResume + resumeMessage = fresh.autoResumeMessage + resumeAt = fresh.resumeAt cwd = fresh.cwd.orEmpty() typedCwd = fresh.cwd.orEmpty() } catch (e: ApiException) { @@ -109,6 +128,8 @@ fun SessionSettingsDialog( // instead of offering a position nothing confirmed. notifyError = e.message notify = null + resumeError = e.message + autoResume = null } } @@ -174,6 +195,39 @@ fun SessionSettingsDialog( } } + /** + * Turns auto-resume on or off, or changes what it would say. + * + * One request for both, because the server takes one: switching it on and typing the message + * are two halves of the same decision, and sending them separately would leave a moment where + * the session is armed with the old words. + * + * Put back if refused, like the notification switch. Turning it off also clears what was + * scheduled -- said here rather than only on the server, or the row would go on naming a time + * that no longer exists. + */ + fun setAutoResume(on: Boolean, message: String) { + val wasOn = autoResume + val wasMessage = resumeMessage + val wasAt = resumeAt + autoResume = on + resumeMessage = message + if (!on) resumeAt = null + resumeError = null + scope.launch { + try { + withContext(Dispatchers.IO) { + setSessionAutoResume(settings, sessionId, on, message) + } + } catch (e: ApiException) { + autoResume = wasOn + resumeMessage = wasMessage + resumeAt = wasAt + resumeError = e.message + } + } + } + // Nothing to do when the name has not changed, so the button says so rather than sending a // request whose success would look exactly like the failure of having typed nothing. val changed = name.trim().isNotEmpty() && name.trim() != title @@ -200,7 +254,10 @@ fun SessionSettingsDialog( onDismissRequest = onDismiss, title = { Text("Session settings") }, text = { - Column { + // Scrollable, because this dialog grew past a screenful: a Material dialog constrains + // its own height and clips what does not fit, so the last control on the list is one + // large system font away from being unreachable with nothing on screen to say so. + Column(Modifier.verticalScroll(rememberScrollState())) { OutlinedTextField( value = name, onValueChange = { name = it }, @@ -244,6 +301,70 @@ fun SessionSettingsDialog( ) } Spacer(Modifier.height(8.dp)) + Row( + verticalAlignment = Alignment.CenterVertically, + modifier = Modifier.fillMaxWidth(), + ) { + Text("Resume after a usage limit", modifier = Modifier.weight(1f)) + if (autoResume == null && resumeError == null) { + CircularProgressIndicator( + modifier = Modifier.width(16.dp).height(16.dp), + strokeWidth = 2.dp, + ) + Spacer(Modifier.width(8.dp)) + } + Switch( + checked = autoResume == true, + onCheckedChange = { setAutoResume(it, resumeMessage) }, + enabled = autoResume != null, + ) + } + // Disabled rather than hidden while the switch is off: a field that comes and goes + // makes its own presence the signal, and a visible one teaches what the switch will + // do. Committed on the keyboard's Done rather than on every keystroke, so typing a + // sentence is one request instead of one per letter. + OutlinedTextField( + value = resumeMessage, + onValueChange = { resumeMessage = it }, + label = { Text("Message to send") }, + // What an empty field means, in the field: the server's own word rather than a + // session poked with nothing to read. + placeholder = { Text(DEFAULT_RESUME_MESSAGE) }, + singleLine = true, + enabled = autoResume == true, + modifier = Modifier.fillMaxWidth(), + keyboardOptions = KeyboardOptions(imeAction = ImeAction.Done), + keyboardActions = + KeyboardActions(onDone = { setAutoResume(true, resumeMessage) }), + ) + // What it does and what it costs, in the order it happens. The last sentence is the + // one that matters: the time below is when the server will *ask*, not a promise + // about when the session speaks. + Text( + "When this session stops because the account is out of quota, the server " + + "checks the limit and sends this message once it has lifted. It checks " + + "again if the limit is still on.", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + // Only where something is actually waiting. Absent is not a state worth a row: a + // session that has not hit a limit has nothing scheduled, which the reader can see + // from the switch. + resumeAt?.let { at -> + Text( + "Waiting now -- next check ${formatCheckTime(at)}.", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } + resumeError?.let { + Text( + it, + color = MaterialTheme.colorScheme.error, + style = MaterialTheme.typography.bodySmall, + ) + } + Spacer(Modifier.height(8.dp)) Row( verticalAlignment = Alignment.CenterVertically, modifier = Modifier.fillMaxWidth(), @@ -399,3 +520,21 @@ fun SessionSettingsDialog( dismissButton = { TextButton(onClick = onDismiss) { Text("Close") } }, ) } + +/** + * When the server will next look, as a local time. + * + * A time rather than a countdown, for the reason the transcript's own limit row gives: this screen + * reads the figure once, and a span drawn from a value nothing refreshes goes stale while somebody + * is looking at it. + */ +private fun formatCheckTime(epochSeconds: Double): String = + try { + DateTimeFormatter.ofLocalizedTime(FormatStyle.SHORT) + .withZone(ZoneId.systemDefault()) + .format(Instant.ofEpochSecond(epochSeconds.toLong())) + } catch (_: Exception) { + // A time that cannot be read is not a time to show: the sentence above still says a check + // is coming, which is the part the reader can act on. + "soon" + } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt index 5089ca2..96eefa1 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt @@ -175,6 +175,18 @@ sealed class TranscriptItem { val preTokens: Long?, val postTokens: Long?, ) : TranscriptItem() + + /** + * The account ran out of quota, so the turn stopped here. + * + * A divider rather than an error: nothing failed, and what a reader scrolling back needs from + * it is the same thing a clear or a compaction gives them -- why the conversation stops at this + * line. + * + * [resetsAt] is epoch seconds and null where the session was told nothing, which is a state the + * row has words for rather than a time it invents. + */ + data class LimitNote(override val seq: Long, val resetsAt: Double?) : TranscriptItem() } /** @@ -458,6 +470,7 @@ fun foldEvent(items: List, entry: SeqEvent): List items + TranscriptItem.LimitNote(entry.seq, event.resetsAt) is SessionEvent.Cleared -> items + TranscriptItem.ClearedNote(entry.seq) is SessionEvent.Compacted -> items + TranscriptItem.CompactedNote(entry.seq, event.preTokens, event.postTokens) diff --git a/app/androidApp/src/test/kotlin/com/example/aiapp/LimitRowTest.kt b/app/androidApp/src/test/kotlin/com/example/aiapp/LimitRowTest.kt new file mode 100644 index 0000000..e4c030f --- /dev/null +++ b/app/androidApp/src/test/kotlin/com/example/aiapp/LimitRowTest.kt @@ -0,0 +1,32 @@ +package com.example.aiapp + +import java.time.ZoneId +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * What the transcript says where a session ran out of quota. + * + * The pair worth a test is the one that reads the same when it goes wrong: a reset time that + * arrived and one that never did. The second must not turn into a plausible-looking time, because a + * reader has no way of telling an invented one from a reported one. + */ +class LimitRowTest { + private val utc = ZoneId.of("UTC") + + @Test + fun `a reported reset time is shown as a time`() { + // 2026-09-05T12:00:00Z. Asserted as a prefix and the clock reading rather than as the + // whole string: the platform's own short-time format is what this asks for, and it + // differs by JDK and locale down to which space character separates the meridiem. + val summary = limitSummary(1_788_609_600.0, utc) + assertTrue(summary.startsWith("Usage limit reached • resets "), summary) + assertTrue(summary.contains("12:00"), summary) + } + + @Test + fun `a limit with no reset time says only what is known`() { + assertEquals("Usage limit reached", limitSummary(null, utc)) + } +} diff --git a/server/src/config.rs b/server/src/config.rs index d699cc8..475a4cb 100644 --- a/server/src/config.rs +++ b/server/src/config.rs @@ -293,6 +293,30 @@ pub struct SessionConfig { /// turned off in one tap where one that never arrived is not diagnosable. #[serde(default = "notify_default")] pub notify: bool, + /// Whether a session stopped by the account's usage limit sends itself a + /// message once the limit lifts, instead of waiting for a person. + /// + /// Off unless somebody asked for it. It spends quota the moment it becomes + /// available and it does so while nobody is looking, which is exactly the + /// kind of thing that must not happen because a default said so. + #[serde(default, skip_serializing_if = "not_set")] + pub auto_resume: bool, + /// What that message says. `None` is [`DEFAULT_RESUME_MESSAGE`], and stays + /// reachable: it is this app's word, not one somebody chose, so clearing + /// the field goes back to it rather than sending an empty message. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub auto_resume_message: Option, + /// The message this session owes itself once the limit lifts, and when to + /// try. Written when a limit is hit, moved when the wait turns out to be + /// wrong, and cleared when the message goes out or auto-resume is turned + /// off -- see [`ScheduledResume`]. + /// + /// Persisted rather than held in memory because the wait outlives the + /// process doing it: a five-hour window and a weekly one both routinely + /// outlast a backend restart, and a resume forgotten across one is a + /// session that silently never comes back. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resume: Option, /// Whether this session's process is stopped when the server exits, instead /// of being left running for the next start to adopt. /// @@ -309,6 +333,28 @@ pub struct SessionConfig { pub created: f64, } +/// A message owed to a session whose account ran out, and when to try sending +/// it. +/// +/// `since` is the whole reason this is a struct: the wait is rescheduled every +/// time the meter is asked and still says no, so `at` alone cannot say how long +/// this has been going on -- and something has to, or a machine that can never +/// be asked is retried until somebody notices. See `crate::resume`. +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ScheduledResume { + /// Epoch seconds: when the limit is next worth checking. Never a promise + /// that the message goes out then -- the meter is asked first. + pub at: f64, + /// Epoch seconds the limit was hit. + pub since: f64, +} + +/// What an auto-resume says when nothing else was chosen. One word, because +/// the session already knows what it was doing and this is only the nudge that +/// lets it carry on. +pub const DEFAULT_RESUME_MESSAGE: &str = "continue"; + fn notify_default() -> bool { true } @@ -486,6 +532,9 @@ mod tests { effort: None, params: BTreeMap::new(), notify: true, + auto_resume: false, + auto_resume_message: None, + resume: None, throwaway: false, created: 1234.5, }], diff --git a/server/src/main.rs b/server/src/main.rs index 9fde3b7..3571b7b 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -16,6 +16,7 @@ mod config; mod files; mod media; mod models; +mod resume; mod routes; mod session; mod setups; @@ -268,6 +269,13 @@ async fn main() -> Result<()> { // that sets it is typed; the monitor is what serves it. let monitor = Arc::new(usage::UsageMonitor::new(manager.usage_fixture())); + // The one thing in here that acts without a request behind it: a session + // switched to auto-resume waits out its account's usage limit and picks + // itself back up. Started whether or not any session has it on, because + // the setting is per session and changes from the phone -- see + // `resume::run`. + tokio::spawn(resume::run(Arc::clone(&manager), Arc::clone(&monitor))); + // The bearer-token middleware wraps the entire router -- routes and fallback // alike -- here and only here, so a new route can't forget auth. let app = routes::router(Arc::clone(&manager)) diff --git a/server/src/resume.rs b/server/src/resume.rs new file mode 100644 index 0000000..2ef56d4 --- /dev/null +++ b/server/src/resume.rs @@ -0,0 +1,382 @@ +//! Auto-resume: picking a session back up when its account's usage limit +//! lifts. +//! +//! Off unless a session was switched to it, because this spends quota the +//! moment quota exists and does it while nobody is watching. What it does is +//! narrow on purpose: it sends one message -- "continue" unless something else +//! was typed -- to a session that stopped because the account ran out, and +//! then it is done. There is no retry loop around the conversation itself. +//! +//! **The schedule is a plan to ask, never a plan to send.** A reset time is +//! the one thing here that cannot be trusted: the dialect's is a hint written +//! when the turn failed, the endpoint's moves when the window moves, and both +//! are wrong across the case this exists for -- a limit that lifts later than +//! it said. So the wait ends in a *question* to [`crate::usage`], and only an +//! answer that says the limits no longer apply sends anything. Every other +//! answer, including one that cannot be got at all, becomes a new wait. +//! +//! This is the top layer: it holds the session manager and the usage monitor +//! and neither holds it. That is what lets the decision below be a pure +//! function of a snapshot and a clock, which is the whole of what is worth +//! testing here. + +use std::sync::Arc; +use std::time::Duration; + +use crate::session::{LimitHit, OwedResume, SessionManager, now}; +use crate::usage::{UsageMonitor, UsageSnapshot, UsageState}; + +/// How often to look at the schedule. Coarse deliberately: a wait measured in +/// hours does not deserve a fine-grained clock, and the meter behind it is +/// cached for three minutes anyway. +const TICK: Duration = Duration::from_secs(60); + +/// How close to a scheduled check is close enough to ask the meter. Anything +/// further out is left alone, so a session waiting five hours costs nothing +/// until the last few minutes of it. +const NEARLY: f64 = 300.0; + +/// How long to wait after an answer that decided nothing -- the machine could +/// not be asked, or it says the limit is still on with no reset time. +const BACKOFF: f64 = 300.0; + +/// The least time to wait before asking again, whatever a reset time says. A +/// window that claims to reset in the past would otherwise be asked about on +/// every tick. +const AT_LEAST: f64 = 60.0; + +/// How long after the limit was hit to stop waiting. +/// +/// Something has to bound it, or a machine that can never be asked -- an +/// unplugged laptop, a setup somebody edited away -- is retried for ever with +/// nothing on screen saying so. A day is past the longest window Claude +/// reports, so reaching this means the wait was never going to end on its own. +const GIVE_UP: f64 = 24.0 * 60.0 * 60.0; + +/// The percentage at which a window is spent. The API counts up to 100, so +/// this is an equality in all but name; written as a threshold because a +/// figure arriving slightly over is a full window, not a corrupt one. +const SPENT: f64 = 100.0; + +/// What to do about one owed resume, having asked the meter. +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum Step { + /// The limits no longer apply: send the message. + Send, + /// Ask again at this epoch second. + WaitUntil(f64), + /// This has been waiting longer than anything real would take. + GiveUp, +} + +/// Runs the schedule until the server stops. +/// +/// Two things wake it: the tick, and a session reporting that it has just run +/// out. The second is not an optimisation -- a limit hit is what *creates* a +/// schedule, and a tick that happened a moment before it would otherwise leave +/// the session unrecorded until the next one. +pub async fn run(manager: Arc, monitor: Arc) { + let mut limits = manager.subscribe_limits(); + loop { + tokio::select! { + _ = tokio::time::sleep(TICK) => {} + hit = limits.recv() => match hit { + Ok(LimitHit { session_id, resets_at }) => note(&manager, &session_id, resets_at), + // Lagged: some reports were dropped, and a session that hit a + // limit while this was busy has no schedule. Nothing is lost + // for good -- the sweep below reads the config, and the + // session will report again the next time it is poked -- but + // it is worth saying, because until then that session waits + // for a person. + Err(tokio::sync::broadcast::error::RecvError::Lagged(missed)) => { + tracing::warn!("auto-resume missed {missed} limit reports"); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + }, + } + sweep(&manager, &monitor).await; + } +} + +/// Records a limit against the session that hit it, if it is one that resumes. +pub(crate) fn note(manager: &SessionManager, session_id: &str, resets_at: Option) { + match manager.note_limit(session_id, resets_at) { + Ok(true) => tracing::info!("session {session_id} hit its usage limit; auto-resume is on"), + Ok(false) => {} + Err(err) => tracing::error!("couldn't schedule a resume for {session_id}: {err:#}"), + } +} + +/// One pass over everything owed a message. +async fn sweep(manager: &SessionManager, monitor: &Arc) { + let at = now(); + for owed in manager.owed_resumes() { + if owed.scheduled.at - at > NEARLY { + continue; + } + // Asked per session rather than once for the whole sweep: the answer + // is cached per machine and per meter, so several sessions on one + // account share one fetch, and a machine nobody is waiting on is not + // dialled at all. + let snapshot = snapshot_for(Arc::clone(monitor), manager, &owed).await; + match decide(snapshot.as_ref(), &owed, now()) { + Step::Send => match manager.resume_now(&owed.session_id) { + Ok(message) => tracing::info!( + "the limit on {} has lifted; sent \"{message}\" to {}", + owed.setup, + owed.session_id + ), + Err(err) => { + tracing::error!("couldn't resume {}: {err:#}", owed.session_id) + } + }, + Step::WaitUntil(next) => { + if let Err(err) = manager.reschedule_resume(&owed.session_id, next) { + tracing::error!( + "couldn't move {}'s resume to {next}: {err:#}", + owed.session_id + ); + } + } + Step::GiveUp => { + // About the machine rather than in the state's own words: the + // detail is in the log, and what lands in the transcript has + // to read on a phone. + let why = match snapshot.as_ref().map(|snapshot| &snapshot.state) { + Some(UsageState::Ok) => "the limit has not lifted in a day".to_string(), + _ => format!("{} could not be asked for a day", owed.setup), + }; + if let Err(err) = manager.abandon_resume(&owed.session_id, &why) { + tracing::error!("couldn't clear {}'s resume: {err:#}", owed.session_id); + } + } + } + } +} + +/// The numbers for the machine and the meter this session is billed against, +/// and `None` when nothing reports on it. +/// +/// Blocking work, so it goes to a blocking thread: the fetch behind it reads a +/// credential file over ssh and then makes an HTTP call. +async fn snapshot_for( + monitor: Arc, + manager: &SessionManager, + owed: &OwedResume, +) -> Option { + let setups: Vec<_> = manager + .setups() + .into_iter() + .filter(|setup| setup.id == owed.setup) + .collect(); + if setups.is_empty() { + return None; + } + let provider = owed.provider; + tokio::task::spawn_blocking(move || { + monitor + .snapshots(&setups) + .into_iter() + .find(|snapshot| snapshot.provider == provider) + }) + .await + .unwrap_or_default() +} + +/// What one owed resume should do, given what the meter said and the time. +/// +/// A pure function of the two, which is what makes the rule inspectable: every +/// answer that is not "the limits no longer apply" is a longer wait, and the +/// only thing that ends the waiting other than success is the clock. +/// +/// The reset time comes from the *snapshot* rather than from the schedule, so +/// a window that turns out to reset later than the dialect said pushes the +/// check back, and one that resets sooner pulls it forward. That is the case +/// the whole design is about: the first answer was a guess, this one is a +/// measurement. +pub fn decide(snapshot: Option<&UsageSnapshot>, owed: &OwedResume, at: f64) -> Step { + let step = match snapshot { + // The meter answered with numbers, which is the only answer that can + // send anything. + Some(snapshot) if snapshot.state == UsageState::Ok => { + let spent: Vec<&crate::usage::UsageWindow> = snapshot + .windows + .iter() + .filter(|window| window.percent >= SPENT) + .collect(); + if spent.is_empty() { + Step::Send + } else { + // The earliest of the spent windows: it is the first moment + // the situation can have changed, and if the others are still + // full this comes straight back here. + match spent + .iter() + .filter_map(|window| epoch_of(window.resets_at.as_deref())) + .min_by(f64::total_cmp) + { + Some(resets) => Step::WaitUntil(resets), + // Spent with no reset time anybody could read. Not a + // reason to send: what is known is that the limit is on. + None => Step::WaitUntil(at + BACKOFF), + } + } + } + // Logged out, unreachable, or the endpoint refused us -- and nothing + // at all, which is a session whose machine or provider has gone. None + // of them says the limit has lifted, and sending on any of them is + // exactly the "inferred value presented as a measured one" this is + // built to avoid. + _ => Step::WaitUntil(at + BACKOFF), + }; + match step { + // Waiting past the point where a real window would have reset means + // whatever is wrong is not going to fix itself. + Step::WaitUntil(_) if at - owed.scheduled.since > GIVE_UP => Step::GiveUp, + Step::WaitUntil(next) => Step::WaitUntil(next.max(at + AT_LEAST)), + other => other, + } +} + +/// An RFC-3339 timestamp as epoch seconds, and `None` for one that is absent +/// or unreadable -- the same two answers the phone's countdown makes, kept +/// apart from each other nowhere here because both mean "this cannot decide +/// when to ask". +fn epoch_of(resets_at: Option<&str>) -> Option { + let text = resets_at?; + time::OffsetDateTime::parse(text, &time::format_description::well_known::Rfc3339) + .ok() + .map(|at| at.unix_timestamp() as f64) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::ScheduledResume; + use crate::usage::UsageWindow; + + fn owed(since: f64) -> OwedResume { + OwedResume { + session_id: "s1".to_string(), + setup: "local".to_string(), + provider: crate::usage::CLAUDE, + scheduled: ScheduledResume { at: since, since }, + } + } + + fn snapshot(state: UsageState, windows: Vec) -> UsageSnapshot { + UsageSnapshot { + provider: crate::usage::CLAUDE.to_string(), + setup: "local".to_string(), + setup_name: "this machine".to_string(), + state, + windows, + fetched_at: 0.0, + } + } + + fn window(percent: f64, resets_at: Option<&str>) -> UsageWindow { + UsageWindow { + kind: "session".to_string(), + label: "5-hour window".to_string(), + percent, + resets_at: resets_at.map(str::to_string), + active: true, + } + } + + #[test] + fn a_meter_with_room_in_it_is_the_only_thing_that_sends() { + let clear = snapshot(UsageState::Ok, vec![window(41.0, None)]); + assert_eq!(decide(Some(&clear), &owed(0.0), 100.0), Step::Send); + } + + #[test] + fn a_window_still_spent_moves_the_check_to_its_own_reset_time() { + // The case the feature exists for: the wait was scheduled for one + // time, the limit is still on, and the endpoint now names another. + let at = 1_788_546_972.0; + let later = "2026-09-05T12:00:00+00:00"; + let spent = snapshot(UsageState::Ok, vec![window(100.0, Some(later))]); + assert_eq!( + decide(Some(&spent), &owed(at - 60.0), at), + Step::WaitUntil(epoch_of(Some(later)).expect("parses")) + ); + } + + #[test] + fn a_reset_time_already_past_still_waits_a_little() { + let at = 1_788_546_972.0; + let spent = snapshot( + UsageState::Ok, + vec![window(100.0, Some("2020-01-01T00:00:00+00:00"))], + ); + assert_eq!( + decide(Some(&spent), &owed(at - 60.0), at), + Step::WaitUntil(at + AT_LEAST) + ); + } + + #[test] + fn the_earliest_spent_window_is_the_one_worth_waiting_on() { + let at = 1_788_546_972.0; + let soon = "2026-09-05T12:00:00+00:00"; + let far = "2026-09-09T12:00:00+00:00"; + let mut weekly = window(100.0, Some(far)); + weekly.kind = "weekly_all".to_string(); + let spent = snapshot(UsageState::Ok, vec![window(100.0, Some(soon)), weekly]); + assert_eq!( + decide(Some(&spent), &owed(at - 60.0), at), + Step::WaitUntil(epoch_of(Some(soon)).expect("parses")) + ); + } + + #[test] + fn a_meter_that_could_not_be_asked_never_sends() { + let at = 1_788_546_972.0; + for state in [ + UsageState::NotLoggedIn, + UsageState::Unreachable { + detail: "no route".to_string(), + }, + UsageState::Failed { + detail: "429".to_string(), + }, + ] { + let broken = snapshot(state.clone(), Vec::new()); + assert_eq!( + decide(Some(&broken), &owed(at - 60.0), at), + Step::WaitUntil(at + BACKOFF), + "{state:?}" + ); + } + // And no snapshot at all -- a machine or provider edited away under a + // session that was waiting on it. + assert_eq!( + decide(None, &owed(at - 60.0), at), + Step::WaitUntil(at + BACKOFF) + ); + } + + #[test] + fn waiting_longer_than_any_real_window_gives_up_rather_than_retrying_for_ever() { + let at = 1_788_546_972.0; + let broken = snapshot( + UsageState::Unreachable { + detail: "no route".to_string(), + }, + Vec::new(), + ); + assert_eq!( + decide(Some(&broken), &owed(at - GIVE_UP - 1.0), at), + Step::GiveUp + ); + // A meter that answers is still allowed to send on the same tick: the + // ceiling bounds waiting, not resuming. + let clear = snapshot(UsageState::Ok, vec![window(3.0, None)]); + assert_eq!( + decide(Some(&clear), &owed(at - GIVE_UP - 1.0), at), + Step::Send + ); + } +} diff --git a/server/src/routes.rs b/server/src/routes.rs index b369c92..b42ff11 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -52,6 +52,8 @@ //! DELETE /sessions/{id} kill process, delete transcript + files //! (?deleteForeign=true removes the machine's own copy too) //! POST /sessions/{id}/notify {notify} -- announce this one or not +//! POST /sessions/{id}/auto-resume {autoResume, message?} -- carry on by itself +//! once the account's usage limit lifts //! GET /notifications SSE: every session's attention-wanting //! moments, live only (see `notifications`) //! GET /defaults {effort} -- what a new session starts at @@ -141,6 +143,7 @@ pub fn router(manager: Arc) -> Router { .route("/sessions/{id}/effort", post(set_effort)) .route("/defaults", get(defaults).post(set_defaults)) .route("/sessions/{id}/notify", post(set_notify)) + .route("/sessions/{id}/auto-resume", post(set_auto_resume)) .route("/notifications", get(notifications)) .route("/sessions/{id}/compact", post(compact)) .route("/sessions/{id}/command", post(command)) @@ -1440,6 +1443,33 @@ async fn set_notify( Ok(StatusCode::NO_CONTENT) } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct AutoResumeRequest { + auto_resume: bool, + /// What to send when the limit lifts. Absent -- and empty, which is what a + /// cleared field sends -- means this app's own default word, which is a + /// choice a caller has to be able to make rather than only start in. + #[serde(default)] + message: Option, +} + +/// Turns auto-resume on or off, and sets what it would say. +/// +/// One request for both, because they are one decision: switching it on +/// without saying what to send is the ordinary case, and changing the words +/// while it is off is how somebody sets it up before it is needed. +async fn set_auto_resume( + State(manager): State>, + UrlPath(id): UrlPath, + axum::Json(body): axum::Json, +) -> Result { + manager + .set_session_auto_resume(&id, body.auto_resume, body.message.as_deref()) + .map_err(bad_request)?; + Ok(StatusCode::NO_CONTENT) +} + #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct CommandRequest { diff --git a/server/src/session/claude/translate.rs b/server/src/session/claude/translate.rs index 3baae30..39c1650 100644 --- a/server/src/session/claude/translate.rs +++ b/server/src/session/claude/translate.rs @@ -224,12 +224,12 @@ impl Translator { .and_then(Value::as_bool) .unwrap_or(false) { - events.push(Event::Error { - message: message - .get("result") - .and_then(Value::as_str) - .unwrap_or("the turn ended with an error") - .to_string(), + let said = message.get("result").and_then(Value::as_str); + events.push(match said.and_then(usage_limit) { + Some(resets_at) => Event::LimitReached { resets_at }, + None => Event::Error { + message: said.unwrap_or("the turn ended with an error").to_string(), + }, }); } let context = self.context.take(); @@ -603,6 +603,37 @@ impl Translator { } } +/// Whether a failed turn failed because the account is out of quota, and when +/// the CLI said the limit lifts. +/// +/// The wording is the CLI's: a turn stopped by the limit ends with `is_error` +/// and a result of `Claude AI usage limit reached|1788546972`, the reset being +/// epoch seconds after a pipe. Matched on the sentence rather than on a code +/// because the CLI sends none, so this is deliberately loose about everything +/// but the four words. +/// +/// The two `None`s mean different things and both are real. The outer one is +/// "some other failure". The inner one is "the limit is reached and the CLI did +/// not say until when" -- which is not a reason to invent a time: `crate::resume` +/// asks the usage endpoint before sending anything, and that answer is the one +/// that decides. +/// +/// Milliseconds are accepted as well as seconds and told apart by magnitude, +/// since a wrong guess would schedule a resume tens of thousands of years out +/// and look exactly like auto-resume being broken. +fn usage_limit(result: &str) -> Option> { + if !result.to_ascii_lowercase().contains("usage limit reached") { + return None; + } + let stamp = result + .rsplit('|') + .next() + .and_then(|tail| tail.trim().parse::().ok()) + .filter(|stamp| *stamp > 0.0) + .map(|stamp| if stamp > 1e11 { stamp / 1000.0 } else { stamp }); + Some(stamp) +} + /// A string field that is there and not empty, or `None`. The CLI omits these /// rather than sending them empty, but a caller that sends `""` means the same /// thing and should not produce a description that draws as a blank line. @@ -1275,6 +1306,55 @@ mod tests { ); } + /// Running out of quota is a state, not a failure of the work. + /// + /// The naive reading -- an error result like any other -- is what shipped + /// before this: the transcript said "Claude AI usage limit reached|…" in + /// red, which is neither readable nor actionable, and nothing above the + /// driver could tell it apart from a broken tool call. + #[test] + fn a_turn_stopped_by_the_usage_limit_says_so_and_carries_the_reset() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Claude AI usage limit reached|1788546972","usage":{}}"#, + ], + ); + assert_eq!( + events[0], + Event::LimitReached { + resets_at: Some(1_788_546_972.0) + } + ); + } + + #[test] + fn a_limit_the_cli_gave_no_reset_for_is_reported_without_one() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut translator = Translator::new(dir.path().to_path_buf()); + let events = translate_lines( + &mut translator, + &[ + r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Claude AI usage limit reached","usage":{}}"#, + ], + ); + // Not a time this side invented: the meter is asked before anything is + // sent, and a made-up reset would only decide when to ask. + assert_eq!(events[0], Event::LimitReached { resets_at: None }); + } + + #[test] + fn a_reset_in_milliseconds_is_not_read_as_the_year_58000() { + assert_eq!( + usage_limit("Claude AI usage limit reached|1788546972000"), + Some(Some(1_788_546_972.0)) + ); + // And anything that is not the limit stays an ordinary failure. + assert_eq!(usage_limit("something broke"), None); + } + /// Pressing Stop is not a failure, and the CLI cannot tell you which it was. /// /// An interrupted turn arrives as exactly the same shape a broken one does, diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index c0e53ad..3af525b 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -302,6 +302,25 @@ pub enum Event { /// it, which is why this is written down rather than left to be inferred /// from a second example that does not exist. Cleared, + /// The account behind this session has no quota left, so the turn stopped + /// without finishing. + /// + /// Its own event rather than an [`Event::Error`] carrying the dialect's + /// sentence, because two things act on it that cannot read English: the + /// transcript draws it as a state the session is in rather than as a + /// failure of something it did, and `crate::resume` schedules the message + /// that picks the work back up. Recognising it belongs to the driver, which + /// is the only layer that knows its dialect's wording -- above here nothing + /// matches on strings. + /// + /// `resets_at` is epoch seconds, and `None` is a real state: the dialect + /// said the limit was hit without saying when it lifts. Nothing here + /// invents one -- what the wait is actually decided against is the usage + /// endpoint, and this is the hint that starts the waiting. + LimitReached { + #[serde(default, skip_serializing_if = "Option::is_none")] + resets_at: Option, + }, Error { message: String, }, diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index 454ee1d..20c2c3f 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -33,6 +33,13 @@ //! `/usage 42 never`, `/usage notloggedin`, `/usage unreachable`, //! `/usage failed`. The vocabulary is `usage::Fixture`'s, where the states //! live. +//! - `/limit [minutes]` -- a turn that stops because the account is out of +//! quota, saying the limit lifts in `minutes` (default 5, and `never` for a +//! limit with no stated reset). What it exists for is auto-resume, which is +//! otherwise reachable only by actually exhausting somebody's account: pair +//! it with `/usage 100 5` for a meter that agrees, and then `/usage 20` for +//! the moment the limit lifts. The wait itself is decided by the meter, so +//! those two commands are the whole rig. //! - `/compact` -- a compaction, start to finish. //! - `/stream N` -- one long answer in N small pieces, 50ms apart: the shape a //! real model's reply arrives in, and the one where the row a reader is @@ -344,6 +351,39 @@ impl EchoDriver { return; } + // A turn that ends the way a real one does when the account runs out: + // the same event a real driver reports, so what acts on it -- the + // transcript row and `crate::resume` -- is exercised rather than + // imitated. The meter it should agree with is `/usage`'s fixture, + // deliberately separate: the two disagreeing is a state worth being + // able to produce, since it is what a stale reset time looks like. + if let Some(rest) = text.strip_prefix("/limit") { + if announce { + self.emit(Event::MessageTaken { + id: None, + text: text.clone(), + attachments, + }); + } + let rest = rest.trim(); + let resets_at = match rest { + "never" | "none" => None, + "" => Some(super::now() + 5.0 * 60.0), + minutes => Some(super::now() + minutes.parse::().unwrap_or(5.0) * 60.0), + }; + self.emit(Event::Status { + state: SessionStatus::Running, + }); + self.emit(Event::AssistantText { + delta: "Working on it".to_string(), + }); + self.emit(Event::LimitReached { resets_at }); + self.emit(Event::Status { + state: SessionStatus::Idle, + }); + return; + } + // The same word the real CLI takes, so a phone drives both the same way. // `Driver::compact` is what the manager's route calls; this is the typed // path onto it. diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index a13601e..463c658 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -28,7 +28,8 @@ use serde::Serialize; use tokio::sync::{broadcast, mpsc}; use crate::config::{ - Config, DriverKind, ProviderConfig, SessionConfig, SetupConfig, SshConfig, TokenEntry, + Config, DEFAULT_RESUME_MESSAGE, DriverKind, ProviderConfig, ScheduledResume, SessionConfig, + SetupConfig, SshConfig, TokenEntry, }; use claude::ClaudeDriver; use driver::{ @@ -49,6 +50,44 @@ const EVENT_BUFFER: usize = 256; /// -- the newest "your turn" is the one still true. const NOTIFICATION_BUFFER: usize = 64; +/// Fan-out buffer for limit reports. One per session per rate-limit window, +/// so a handful a day across everything -- but sized like the notifications +/// above rather than at 1, because the only subscriber is a task that may be +/// mid-tick when several arrive. +const LIMIT_BUFFER: usize = 64; + +/// How long after a limit with no stated reset to ask the meter about it. +/// Short, because the meter is the authority and this is only how soon it is +/// worth the first question. +const FIRST_CHECK: f64 = 60.0; + +/// A session that stopped because its account is out of quota, as the pump +/// saw it. +/// +/// Broadcast downward rather than acted on here, for the reason `Shared` +/// gives: the pump runs underneath the manager and reaching back up would +/// invert that. `crate::resume` is the one subscriber, and what it does with +/// this is decided by the session's own `auto_resume`. +/// The two channels a pump reports on, which carry what this layer records +/// but does not act on: what a phone should be told, and what +/// `crate::resume` should schedule. +/// +/// One struct because they travel together through every launch and every +/// pump, and a second one arriving should not be a third parameter on both. +#[derive(Clone)] +pub struct Announcements { + notifications: broadcast::Sender, + limits: broadcast::Sender, +} + +#[derive(Debug, Clone)] +pub struct LimitHit { + pub session_id: String, + /// Epoch seconds the dialect said the limit lifts, and `None` where it + /// said nothing. Only ever a hint -- see [`Event::LimitReached`]. + pub resets_at: Option, +} + pub fn now() -> f64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -96,6 +135,54 @@ pub enum NotificationKind { Finished, } +/// What a session's auto-resume setting looks like from outside: on or off, +/// what it would say, and when it next intends to check. +/// +/// One struct rather than three parameters on [`LiveSession::info`], and read +/// from the config rather than from the launch snapshot beside it, for the +/// reason `cwd` is: all three change under a running session. +#[derive(Debug, Clone)] +pub struct AutoResumeView { + pub on: bool, + pub message: String, + pub at: Option, +} + +impl AutoResumeView { + fn of(meta: &SessionConfig) -> Self { + Self { + on: meta.auto_resume, + message: resume_message(meta), + at: meta.resume.map(|scheduled| scheduled.at), + } + } +} + +/// A session with a message owed to it once its account has quota again -- +/// see [`SessionManager::owed_resumes`]. +/// +/// Carries no message: what to send is read under the lock at the moment it is +/// sent (see [`SessionManager::resume_now`]), because a wait lasts hours and +/// the words can be edited from the phone inside one. +#[derive(Debug, Clone)] +pub struct OwedResume { + pub session_id: String, + /// The machine whose account ran out, which is the one to ask. + pub setup: String, + /// Which meter reports on it -- a `crate::usage::UsageProvider::name`, the + /// same pairing `SessionInfo::usage_provider` uses. + pub provider: &'static str, + pub scheduled: ScheduledResume, +} + +/// What a session's auto-resume says, with the default filled in. One place, +/// so the phone is shown the words that would actually be sent. +fn resume_message(meta: &SessionConfig) -> String { + meta.auto_resume_message + .clone() + .unwrap_or_else(|| DEFAULT_RESUME_MESSAGE.to_string()) +} + /// One row of `GET /sessions`. #[derive(Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] @@ -162,6 +249,19 @@ pub struct SessionInfo { /// reason `permission_mode` is: a switch that guesses its own position /// is how you turn something off while believing you are reading it. pub notify: bool, + /// Whether this session sends itself a message when its account's usage + /// limit lifts, and what that message says. Reported for the same reason + /// `notify` is. + pub auto_resume: bool, + /// The words that would be sent, with the default already filled in -- + /// the phone shows what would actually happen rather than an empty field + /// meaning "something". + pub auto_resume_message: String, + /// Epoch seconds this session next intends to check whether the limit has + /// lifted, and absent when nothing is waiting. A measurement rather than + /// a promise: what decides is the meter, asked at that moment. + #[serde(skip_serializing_if = "Option::is_none")] + pub resume_at: Option, pub status: SessionStatus, pub last_activity: f64, pub created: f64, @@ -450,6 +550,7 @@ impl LiveSession { effort: Option<&str>, imported: bool, kind: Option, + resume: AutoResumeView, ) -> SessionInfo { SessionInfo { id: self.meta.id.clone(), @@ -466,6 +567,9 @@ impl LiveSession { takes_effort: kind.is_some_and(DriverKind::takes_effort), context_tokens: *self.shared.context_tokens.lock().unwrap(), notify: *self.shared.notify.lock().unwrap(), + auto_resume: resume.on, + auto_resume_message: resume.message, + resume_at: resume.at, max_image_edge: kind.and_then(DriverKind::max_image_edge), usage_provider: kind.and_then(DriverKind::usage_provider), imported, @@ -491,8 +595,9 @@ pub struct SessionManager { /// Downloaded GGUF models, shared by every session that names one, /// which is why they sit beside the session directories. models_dir: PathBuf, - /// Where every session's pump sends what a phone should be told about. - notifications: broadcast::Sender, + /// Where every session's pump reports what this layer does not act on -- + /// see [`Announcements`]. + announce: Announcements, /// Imports and deletes running against a machine's Claude Code /// sessions: like the notifications, state the phone reads but does not /// own. @@ -520,6 +625,11 @@ impl SessionManager { wg_app_link::private::create_dir(&data_dir)?; let (notifications, _) = broadcast::channel(NOTIFICATION_BUFFER); + let (limits, _) = broadcast::channel(LIMIT_BUFFER); + let announce = Announcements { + notifications, + limits, + }; // Made here rather than passed in, and handed *out* to the usage // monitor by whoever wires the two together: every echo driver // this manager builds gets a clone, including the ones built @@ -540,7 +650,7 @@ impl SessionManager { models_dir: &models_dir, usage: &usage_fixture, }, - notifications.clone(), + announce.clone(), // Nothing is started here; see `Launching`. Launching::Restart, ) @@ -557,7 +667,7 @@ impl SessionManager { config_path, data_dir, models_dir, - notifications, + announce, pending: Arc::new(pending::Registry::default()), spawn_throwaway: false, usage_fixture, @@ -923,6 +1033,7 @@ impl SessionManager { meta.effort.as_deref(), import::read_cursor(&self.data_dir.join(&meta.id)).is_some(), kind_of(&inner.config, &meta.setup, &meta.provider), + AutoResumeView::of(meta), ), None => SessionInfo { id: meta.id.clone(), @@ -941,6 +1052,9 @@ impl SessionManager { usage_provider: kind_of(&inner.config, &meta.setup, &meta.provider) .and_then(DriverKind::usage_provider), notify: meta.notify, + auto_resume: meta.auto_resume, + auto_resume_message: resume_message(meta), + resume_at: meta.resume.map(|scheduled| scheduled.at), imported: import::read_cursor(&self.data_dir.join(&meta.id)).is_some(), keeps_own_transcript: keeps_own_transcript( &inner.config, @@ -959,8 +1073,15 @@ impl SessionManager { /// Every session's attention-wanting moments, on one stream. One /// connection for the whole backend rather than one per session: the /// phone subscribes while showing no session at all. + /// Every session running out of quota, on one stream -- the other half of + /// [`SessionManager::owed_resumes`]. Subscribed to by `crate::resume`, so + /// a limit hit is acted on when it happens rather than at the next tick. + pub fn subscribe_limits(&self) -> broadcast::Receiver { + self.announce.limits.subscribe() + } + pub fn subscribe_notifications(&self) -> broadcast::Receiver { - self.notifications.subscribe() + self.announce.notifications.subscribe() } /// Imports and deletes running against importable sessions -- see @@ -1055,6 +1176,12 @@ impl SessionManager { // On by default. Not offered at spawn: a session's first turn // is exactly the one somebody is waiting for. notify: true, + // Off, and not offered at spawn either -- for the opposite + // reason: this one spends quota with nobody watching, so it is + // asked for on a session somebody already has, never inherited. + auto_resume: false, + auto_resume_message: None, + resume: None, // Recorded on the session rather than remembered here, so // whichever server is running when the time comes knows what to // do with it -- see `SessionConfig::throwaway`. @@ -1067,7 +1194,7 @@ impl SessionManager { &setup, &provider, self.env(), - self.notifications.clone(), + self.announce.clone(), Launching::Asked(seed), )?; let mut candidate = inner.config.clone(); @@ -1088,6 +1215,7 @@ impl SessionManager { session.meta.effort.as_deref(), import::read_cursor(&self.data_dir.join(&id)).is_some(), Some(provider.kind), + AutoResumeView::of(&session.meta), ); inner.live.insert(id, session); Ok(info) @@ -1150,6 +1278,176 @@ impl SessionManager { Ok(()) } + /// Turns auto-resume on or off for one session, and sets what it will + /// say. + /// + /// Turning it off cancels anything already scheduled, which is the path + /// out of the state the previous call put the session in: a message left + /// owed by a switch somebody has since turned off would arrive hours + /// later with nothing on screen to explain it. + /// + /// An empty message is not a message -- it is what a cleared field sends + /// -- so it means [`DEFAULT_RESUME_MESSAGE`] rather than a session poked + /// with nothing to read. + pub fn set_session_auto_resume( + &self, + id: &str, + auto_resume: bool, + message: Option<&str>, + ) -> Result<()> { + let mut inner = self.inner.write().unwrap(); + if !inner.config.sessions.iter().any(|meta| meta.id == id) { + bail!("no session {id}"); + } + let mut candidate = inner.config.clone(); + for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { + meta.auto_resume = auto_resume; + meta.auto_resume_message = message + .map(str::trim) + .filter(|message| !message.is_empty()) + .map(str::to_string); + if !auto_resume { + meta.resume = None; + } + } + candidate.save(&self.config_path)?; + inner.config = candidate; + Ok(()) + } + + /// Records that a session ran out of quota, and when to look again. + /// + /// Does nothing for a session that does not auto-resume, and nothing for + /// one already waiting: a turn that fails twice against the same window + /// is the same wait, and taking the second report would push the check + /// back every time the session was poked. + /// + /// `resets_at` is the dialect's hint and is used only to decide when to + /// *ask*; [`crate::resume`] asks the meter before anything is sent. A + /// session told nothing is checked shortly, since the meter is the + /// authority either way. + pub fn note_limit(&self, id: &str, resets_at: Option) -> Result { + let mut inner = self.inner.write().unwrap(); + let meta = inner + .config + .sessions + .iter() + .find(|meta| meta.id == id) + .with_context(|| format!("no session {id}"))?; + if !meta.auto_resume || meta.resume.is_some() { + return Ok(false); + } + let at = now(); + let scheduled = ScheduledResume { + at: resets_at.unwrap_or(at + FIRST_CHECK), + since: at, + }; + let mut candidate = inner.config.clone(); + for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { + meta.resume = Some(scheduled); + } + candidate.save(&self.config_path)?; + inner.config = candidate; + Ok(true) + } + + /// Every session with a message owed to it, oldest schedule first. + /// + /// Carries what deciding needs rather than a session id to look things up + /// by, so the scheduler holds no lock while it makes a network call: the + /// machine and the meter to ask, and the words to send. + pub fn owed_resumes(&self) -> Vec { + let inner = self.inner.read().unwrap(); + let mut owed: Vec = inner + .config + .sessions + .iter() + .filter_map(|meta| { + let scheduled = meta.resume?; + Some(OwedResume { + session_id: meta.id.clone(), + setup: meta.setup.clone(), + provider: kind_of(&inner.config, &meta.setup, &meta.provider)? + .usage_provider()?, + scheduled, + }) + }) + .collect(); + owed.sort_by(|a, b| a.scheduled.at.total_cmp(&b.scheduled.at)); + owed + } + + /// Moves a scheduled check later (or earlier), leaving everything else + /// about it alone -- including when the limit was hit, which is what + /// bounds the retrying. + pub fn reschedule_resume(&self, id: &str, at: f64) -> Result<()> { + let mut inner = self.inner.write().unwrap(); + let mut candidate = inner.config.clone(); + for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { + if let Some(scheduled) = meta.resume.as_mut() { + scheduled.at = at; + } + } + candidate.save(&self.config_path)?; + inner.config = candidate; + Ok(()) + } + + /// Sends the message this session is owed and clears the schedule. + /// + /// Cleared first, and saved before the message goes out: a send that + /// fails leaves nothing owed, where a schedule left standing by a failed + /// send is one that fires again on the next tick and every tick after. + /// The session is started if it has none, exactly as any other message + /// does. + pub fn resume_now(&self, id: &str) -> Result { + let message = { + let mut inner = self.inner.write().unwrap(); + let meta = inner + .config + .sessions + .iter() + .find(|meta| meta.id == id) + .with_context(|| format!("no session {id}"))?; + let message = resume_message(meta); + let mut candidate = inner.config.clone(); + for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { + meta.resume = None; + } + candidate.save(&self.config_path)?; + inner.config = candidate; + message + }; + self.send_message(id, message.clone(), Vec::new())?; + Ok(message) + } + + /// Gives up on a scheduled resume, and says so in the transcript. + /// + /// In the transcript because that is where somebody looking at this + /// session will be: a wait that quietly stopped waiting is + /// indistinguishable from one still going, and the session is sitting + /// there having said nothing since the limit was hit. + pub fn abandon_resume(&self, id: &str, why: &str) -> Result<()> { + { + let mut inner = self.inner.write().unwrap(); + let mut candidate = inner.config.clone(); + for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) { + meta.resume = None; + } + candidate.save(&self.config_path)?; + inner.config = candidate; + } + if let Some(session) = self.session(id) { + let _ = session.sink.send(Event::Error { + message: format!( + "auto-resume gave up on this session: {why}. Send it something to carry on." + ), + }); + } + Ok(()) + } + /// Renames a session: persisted, shown, and passed on to whatever is /// running it. /// @@ -1540,7 +1838,7 @@ impl SessionManager { &setup, &provider, self.env(), - self.notifications.clone(), + self.announce.clone(), Launching::Asked(None), )?; inner.live.insert(id.to_string(), session); @@ -1935,7 +2233,7 @@ fn launch( setup: &SetupConfig, provider: &ProviderConfig, env: Env<'_>, - notifications: broadcast::Sender, + announce: Announcements, why: Launching, ) -> Result> { let dir = env.data_dir.join(&meta.id); @@ -2069,7 +2367,7 @@ fn launch( Arc::clone(&shared), events.clone(), Arc::clone(&commands), - notifications, + announce, )); Ok(Arc::new(LiveSession { @@ -2191,7 +2489,7 @@ async fn pump( shared: Arc, events: broadcast::Sender, commands: Arc, - notifications: broadcast::Sender, + announce: Announcements, ) { // Messages the session has been given and not started reading, which is // what makes a turn ending not the same thing as the work ending. @@ -2270,7 +2568,7 @@ async fn pump( { // No subscribers is the ordinary case -- nobody has // the app open -- and it is not an error. - let _ = notifications.send(Notification { + let _ = announce.notifications.send(Notification { session_id: id.clone(), title: shared.title.lock().unwrap().clone(), kind, @@ -2278,6 +2576,16 @@ async fn pump( }); } } + if let Event::LimitReached { resets_at } = &entry.event { + // Sent whether or not this session auto-resumes: whether + // to act is the manager's decision, and it is the one + // holding the setting. No subscribers is the ordinary + // case -- nothing waits on this in the tests. + let _ = announce.limits.send(LimitHit { + session_id: id.clone(), + resets_at: *resets_at, + }); + } *shared.last_activity.lock().unwrap() = ts; *shared.written.lock().unwrap() += 1; // The turn's own first line, kept for whatever arrives at the @@ -2631,6 +2939,99 @@ mod tests { ); } + /// The whole of the server's half of auto-resume, driven by echo: a + /// limit is reported, the session that asked for it is scheduled, and the + /// one that did not is left alone. + /// + /// Echo rather than the Claude CLI on purpose -- reaching this state for + /// real means exhausting an account, and the event both drivers report is + /// the same one. + #[tokio::test] + async fn a_limit_schedules_a_resume_only_where_one_was_asked_for() { + let dir = tempfile::tempdir().expect("tempdir"); + let config_path = dir.path().join("config.ron"); + let data_dir = dir.path().join("sessions"); + seed_echo_only(&config_path); + let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) + .expect("manager"); + let quiet = manager.spawn_session(echo_spec()).expect("spawn"); + let resuming = manager.spawn_session(echo_spec()).expect("spawn"); + manager + .set_session_auto_resume(&resuming.id, true, Some("carry on")) + .expect("on"); + + let mut limits = manager.subscribe_limits(); + for id in [&quiet.id, &resuming.id] { + manager + .session(id) + .expect("live") + .send_message("/limit 10".to_string(), Vec::new()); + } + // Both report; only one is owed anything. Drained rather than slept + // through, so the assertions below cannot run before the events they + // are about. + for _ in 0..2 { + let hit = tokio::time::timeout(Duration::from_secs(5), limits.recv()) + .await + .expect("a limit within five seconds") + .expect("channel open"); + crate::resume::note(&manager, &hit.session_id, hit.resets_at); + } + + let owed = manager.owed_resumes(); + assert_eq!( + owed.iter().map(|owed| &owed.session_id).collect::>(), + vec![&resuming.id], + "a session nobody switched on was scheduled anyway" + ); + // The dialect's hint decides when to *ask*, so it is what was written + // down -- ten minutes out, not the minute a session told nothing gets. + assert!( + owed[0].scheduled.at - now() > FIRST_CHECK, + "the reset time the session reported was ignored" + ); + + // Turning it off is the way out of the state turning it on created. + manager + .set_session_auto_resume(&resuming.id, false, None) + .expect("off"); + assert!( + manager.owed_resumes().is_empty(), + "a message stayed owed after auto-resume was switched off" + ); + } + + /// What the phone reads back, which is what its switch and its text field + /// are drawn from. + #[tokio::test] + async fn a_session_reports_its_auto_resume_setting_and_its_default_words() { + let dir = tempfile::tempdir().expect("tempdir"); + let config_path = dir.path().join("config.ron"); + let data_dir = dir.path().join("sessions"); + seed_echo_only(&config_path); + let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) + .expect("manager"); + let info = manager.spawn_session(echo_spec()).expect("spawn"); + assert!(!info.auto_resume); + // The default is reported rather than left empty: the field shows + // what would actually be sent. + assert_eq!(info.auto_resume_message, DEFAULT_RESUME_MESSAGE); + assert_eq!(info.resume_at, None); + + // An empty message is what a cleared field sends, and means the + // default rather than a session poked with nothing to read. + manager + .set_session_auto_resume(&info.id, true, Some(" ")) + .expect("on"); + let fresh = manager + .sessions() + .into_iter() + .find(|session| session.id == info.id) + .expect("listed"); + assert!(fresh.auto_resume); + assert_eq!(fresh.auto_resume_message, DEFAULT_RESUME_MESSAGE); + } + /// The switch reaches the running pump, not just the config file. The /// failure is silent in the direction that matters: a /// `set_session_notify(false)` writing only the config looks correct on @@ -2659,7 +3060,20 @@ mod tests { // open to look one up on. assert_eq!( first.title, - session.info("m", None, None, false, None).title + session + .info( + "m", + None, + None, + false, + None, + AutoResumeView { + on: false, + message: DEFAULT_RESUME_MESSAGE.to_string(), + at: None, + }, + ) + .title ); manager.set_session_notify(&info.id, false).expect("off");