iris/ai-app
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main'

Browse Source
This commit is contained in:
iris committed 2026-08-30 00:14:23 -04:00
2 parents 50670863f4 2f4dff1435
commit 49d3439ec4
22 files changed
+1118 -54
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/androidApp/src/main/AndroidManifest.xml
+25
View File
@@ -10,6 +10,14 @@
dev-updater's manifest for the full story. -->
<uses-permission android:name="android.permission.ACCESS_LOCAL_NETWORK" />
<!-- Telling somebody a session wants them. POST_NOTIFICATIONS is a
runtime permission from Android 13; the foreground-service pair
below is what lets the connection outlive the app being closed,
which is the entire point (see Notifications.kt). -->
<uses-permission android:name="android.permission.POST_NOTIFICATIONS" />
<uses-permission android:name="android.permission.FOREGROUND_SERVICE" />
<uses-permission android:name="android.permission.FOREGROUND_SERVICE_SPECIAL_USE" />
<!-- tools:ignore MissingApplicationIcon: there is no icon yet, and
that is a decision rather than an oversight. An app with no icon
of its own is obvious to anyone who opens a launcher, so the
@@ -49,6 +57,23 @@
</intent-filter>
</activity>
<!-- specialUse rather than dataSync, which is the type this looks
like: Android 15 caps dataSync at six hours a day, and a
connection that stops listening after six hours is one that
misses the overnight run it exists for. The subtype below is
the reason string that type requires. -->
<service
android:name=".NotificationService"
android:exported="false"
android:foregroundServiceType="specialUse">
<property
android:name="android.app.PROPERTY_SPECIAL_USE_FGS_SUBTYPE"
android:value="Holds one connection to the user's own backend so a session
that needs an answer can be reported while the app is closed. There is no
push service: the backend is reachable only over the user's WireGuard
tunnel and never talks to a third party." />
</service>
<!-- The scanner behind Settings' "Scan QR code". Declared here so
it can drop the library CaptureActivity's landscape pin: the
code being scanned is usually on a monitor in front of someone
app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt
+35
View File
@@ -141,6 +141,18 @@ data class SessionSummary(
* Whether this continues a session the machine already had, which changes what deleting means.
*/
val imported: Boolean,
/**
* Whether this session announces itself when it wants attention.
*
* Reported rather than assumed, for the same reason [permissionMode] is: a switch that draws
* itself from a default is one you can turn off while believing you are reading it. Defaults to
* on when a backend is too old to say, which matches what that backend actually does.
*/
val notify: Boolean,
/**
* Every token this session has spent, as the server counts it -- see `SessionEvent.UsageDelta`.
*/
val totalTokens: Long,
val status: String,
val lastActivity: Double,
)
@@ -156,6 +168,8 @@ private fun parseSession(session: JSONObject) =
model = session.optString("model").ifEmpty { null },
permissionMode = session.optString("permissionMode").ifEmpty { null },
imported = session.optBoolean("imported", false),
notify = session.optBoolean("notify", true),
totalTokens = session.optLong("totalTokens", 0),
status = session.getString("status"),
lastActivity = session.getDouble("lastActivity"),
)
@@ -163,6 +177,17 @@ private fun parseSession(session: JSONObject) =
fun fetchSessions(settings: ServerSettings): List<SessionSummary> =
requestFromServer(settings, "/sessions") { it.jsonObjects(::parseSession) }
/**
* One session as the server has it now.
*
* For screens whose controls are *set to* something rather than merely showing it. A screen opened
* from a list row carries the row the list last fetched, which is a snapshot: fine for a title,
* wrong for a switch, since a switch drawn from a stale row shows a position that may have been
* changed since -- here or on another device -- and nothing on screen says which.
*/
fun fetchSession(settings: ServerSettings, sessionId: String): SessionSummary =
requestFromServer(settings, "/sessions/$sessionId") { parseSession(it.jsonObject()) }
// What the server offers, so the spawn screen has no hardcoded lists: a
// setup added to the server's config.ron appears here with no app rebuild.
//
@@ -588,6 +613,16 @@ fun setSessionPermissionMode(settings: ServerSettings, sessionId: String, mode:
) {}
}
/** Turns this session's notifications on or off. Stored on the backend -- see `SessionConfig`. */
fun setSessionNotify(settings: ServerSettings, sessionId: String, notify: Boolean) {
requestFromServer(
settings,
"/sessions/$sessionId/notify",
method = "POST",
jsonBody = JSONObject().put("notify", notify).toString(),
) {}
}
/**
* Asks the session to run one of its own commands.
*
app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt
+11 -2
View File
@@ -102,7 +102,15 @@ sealed class SessionEvent {
*/
data class Settings(val model: String?, val permissionMode: String?) : SessionEvent()
data class UsageDelta(val tokens: Long) : SessionEvent()
/**
* What a turn cost, and what the session has cost in total.
*
* [total] is the server's running figure, carried on the event so a reader never adds up its
* own: a phone opens a session on the newest page of the transcript, so a sum it computed would
* be that page's share of the conversation wearing the whole conversation's label. Zero on
* entries recorded before the backend sent it.
*/
data class UsageDelta(val tokens: Long, val total: Long) : SessionEvent()
/**
* A compaction that finished, and how much context it recovered.
@@ -200,7 +208,8 @@ fun parseSeqEvent(json: String): SeqEvent {
model = body.optString("model").ifEmpty { null },
permissionMode = body.optString("permissionMode").ifEmpty { null },
)
"usageDelta" -> SessionEvent.UsageDelta(body.getLong("tokens"))
"usageDelta" ->
SessionEvent.UsageDelta(body.getLong("tokens"), body.optLong("total", 0))
"compacted" ->
SessionEvent.Compacted(
preTokens = if (body.has("preTokens")) body.getLong("preTokens") else null,
app/androidApp/src/main/kotlin/com/example/aiapp/MainActivity.kt
+21
View File
@@ -33,6 +33,16 @@ class MainActivity : ComponentActivity() {
private val requestLocalNetworkPermission =
registerForActivityResult(ActivityResultContracts.RequestPermission()) {}
/**
* The service starts either way, and posts nothing if this is refused.
*
* Deliberately not gated on the answer: the permission can be granted later from Android's own
* settings, and a service that only ever started at the moment it was granted would then stay
* down until the app was launched again -- which is the case notifications exist to avoid.
*/
private val requestNotificationPermission =
registerForActivityResult(ActivityResultContracts.RequestPermission()) {}
override fun onCreate(savedInstanceState: Bundle?) {
super.onCreate(savedInstanceState)
@@ -55,7 +65,15 @@ class MainActivity : ComponentActivity() {
requestLocalNetworkPermission.launch(Manifest.permission.ACCESS_LOCAL_NETWORK)
}
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.TIRAMISU) {
requestNotificationPermission.launch(Manifest.permission.POST_NOTIFICATIONS)
}
handleEnrollment(intent)
// After enrollment, so a first launch that arrives with a token
// starts the service with something to connect to rather than
// stopping it and waiting for the next launch.
NotificationService.sync(this)
setContent {
MaterialTheme(colorScheme = AiAppColors) {
@@ -94,6 +112,9 @@ class MainActivity : ComponentActivity() {
}
saveServerSettings(this, settings)
settingsVersion++
// Enrolling is the moment there is a backend to watch, and
// re-enrolling elsewhere is the moment the old one stops being it.
NotificationService.sync(this)
Toast.makeText(this, "Enrolled with ${settings.baseUrl}", Toast.LENGTH_LONG).show()
}
}
app/androidApp/src/main/kotlin/com/example/aiapp/Markdown.kt
+13 -6
View File
@@ -59,21 +59,28 @@ fun MarkdownText(text: String, modifier: Modifier = Modifier) {
bullet = body,
list = body,
table = body,
// Code in a monospace face: a code block set in the body font stops looking like
// code at all. The colour rides on the style here rather than in `markdownColor`,
// which stopped carrying `codeText`/`inlineCodeText`/`linkText` when the renderer
// moved them onto the typography.
// Code in a monospace face, in the ordinary text colour. The face and the tinted
// background are what say "this is code"; colour is not, and it used to be green
// -- the palette's colour for a *literal*. A block of code is not a literal, it
// is text that happens to be code, and painting all of it green said the whole
// block was one. Where a literal really does appear inside code, the thing that
// should colour it is a syntax highlighter looking at the code, which is exactly
// what a tool call's input already gets from `catppuccinSyntax`.
//
// The colour rides on the style here rather than in `markdownColor`, which
// stopped carrying `codeText`/`inlineCodeText`/`linkText` when the renderer moved
// them onto the typography.
code =
MaterialTheme.typography.bodyMedium.copy(
fontFamily = FontFamily.Monospace,
color = codeColor,
color = MaterialTheme.colorScheme.onSurface,
),
inlineCode =
body.copy(
fontFamily = FontFamily.Monospace,
// Unspecified so an inline span keeps the size of the line it sits in.
fontSize = TextUnit.Unspecified,
color = codeColor,
color = MaterialTheme.colorScheme.onSurface,
),
textLink =
TextLinkStyles(
app/androidApp/src/main/kotlin/com/example/aiapp/Notifications.kt
+273
View File
@@ -0,0 +1,273 @@
package com.example.aiapp
import android.Manifest
import android.app.Notification
import android.app.PendingIntent
import android.app.Service
import android.content.Context
import android.content.Intent
import android.content.pm.PackageManager
import android.content.pm.ServiceInfo
import android.os.Build
import android.os.IBinder
import androidx.core.app.NotificationChannelCompat
import androidx.core.app.NotificationCompat
import androidx.core.app.NotificationManagerCompat
import androidx.core.app.ServiceCompat
import androidx.core.content.ContextCompat
import java.io.IOException
import java.net.HttpURLConnection
import java.net.URL
import kotlin.concurrent.thread
import org.json.JSONObject
/**
* Telling somebody a session wants them, when they are not looking at the app.
*
* This is a **foreground service**, which on Android is the only way to keep a connection open
* while the app is closed -- there has been no such thing as a long-lived background service since
* Android 8. It is what Syncthing does for the same reason. Discord is not a counter-example: it
* gets a push from Google's servers, which would mean this backend talking to Google about
* somebody's coding sessions, and the whole point of the tunnel is that it does not.
*
* The cost Android charges for it is a notification of its own that cannot be dismissed. That is
* made as quiet as the platform allows: [ONGOING_CHANNEL] is `IMPORTANCE_MIN`, so it makes no
* sound, shows no status-bar icon, and sits at the bottom of the shade -- the same arrangement
* Syncthing's "hide the persistent notification" option produces. It is not hidden outright,
* because it cannot be and because it should not be: it is the honest indicator that something is
* holding a connection open.
*/
class NotificationService : Service() {
@Volatile private var stream: HttpURLConnection? = null
@Volatile private var stopping = false
override fun onBind(intent: Intent?): IBinder? = null
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
val settings = loadServerSettings(this)
if (settings == null) {
// Nothing to connect to. Stopping rather than idling: a service
// holding no connection still costs the ongoing notification,
// which would then be announcing work that is not happening.
stopSelf()
return START_NOT_STICKY
}
// Through ServiceCompat so the type is stated once and ignored on
// the versions that predate types, rather than branching here.
ServiceCompat.startForeground(this, ONGOING_ID, ongoingNotification(), foregroundType())
thread(isDaemon = true, name = "ai-app-notifications") { follow(settings) }
// Restarted if Android kills it, which is the whole point: the
// window this covers is exactly the one where nobody is watching.
return START_STICKY
}
override fun onDestroy() {
stopping = true
stream?.disconnect()
}
/**
* Follows the backend's notification stream, reconnecting until stopped.
*
* A dropped connection is the ordinary case here rather than an error -- a phone changes
* networks, the tunnel comes and goes, the backend restarts -- so it retries quietly and
* forever. Nothing is shown when it cannot connect: a notification saying "I could not tell you
* whether anything happened" on a phone in somebody's pocket is noise about a condition they
* cannot act on, and the session list already says what is waiting when they next look.
*/
private fun follow(settings: ServerSettings) {
while (!stopping) {
try {
readStream(settings)
} catch (_: IOException) {
// Deliberate: see above.
}
if (stopping) return
try {
Thread.sleep(RECONNECT_DELAY_MS)
} catch (_: InterruptedException) {
return
}
}
}
private fun readStream(settings: ServerSettings) {
val connection =
URL("${settings.baseUrl}/notifications").openConnection() as HttpURLConnection
stream = connection
try {
connection.applyPinnedTls()
connection.connectTimeout = CONNECT_TIMEOUT_MS
// No read timeout, for the reason EventStream gives: between
// notifications there is nothing to read, possibly for hours.
connection.readTimeout = 0
connection.setRequestProperty("Authorization", "Bearer ${settings.token}")
connection.setRequestProperty("Accept", "text/event-stream")
if (connection.responseCode != 200) {
throw IOException("HTTP ${connection.responseCode} for the notification stream")
}
val reader = connection.inputStream.bufferedReader()
val data = StringBuilder()
while (!stopping) {
val line = reader.readLine() ?: break
when {
line.isEmpty() -> {
if (data.isNotEmpty()) show(parseNotification(data.toString()))
data.clear()
}
line.startsWith("data:") -> data.append(line.removePrefix("data:").trim())
else -> {} // comments (keep-alives) and ids: nothing to do
}
}
} finally {
connection.disconnect()
stream = null
}
}
/**
* One notification per session, replacing that session's previous one.
*
* Keyed by session id rather than accumulating: two sessions wanting attention are two things
* to know about, but one session that finished and then asked a question is one thing -- the
* question. A stack of stale rows for the same conversation is how a notification drawer
* becomes something to clear rather than read.
*/
private fun show(notification: SessionNotification) {
val manager = NotificationManagerCompat.from(this)
// Two different noes, and both are answers rather than faults: the runtime permission
// refused, and notifications switched off for the app in Android's own settings. Neither
// is reported anywhere -- the person said no, and saying it back to them through the
// channel they closed is not available anyway.
val allowed =
ContextCompat.checkSelfPermission(this, Manifest.permission.POST_NOTIFICATIONS) ==
PackageManager.PERMISSION_GRANTED
if (!allowed || !manager.areNotificationsEnabled()) {
return
}
val open =
PendingIntent.getActivity(
this,
0,
Intent(this, MainActivity::class.java)
.setAction(Intent.ACTION_MAIN)
.addCategory(Intent.CATEGORY_LAUNCHER),
PendingIntent.FLAG_UPDATE_CURRENT or PendingIntent.FLAG_IMMUTABLE,
)
val built =
NotificationCompat.Builder(this, ALERT_CHANNEL)
.setContentTitle(notification.title)
.setContentText(
when (notification.kind) {
// What the reader has to do, not what the session
// did: "awaitingInput" is the wire's word and says
// nothing to somebody reading a lock screen.
"awaitingInput" -> "Waiting for you"
else -> "Finished"
}
)
.setSmallIcon(android.R.drawable.stat_notify_chat)
.setContentIntent(open)
.setAutoCancel(true)
.setWhen((notification.at * 1000).toLong())
.setShowWhen(true)
.build()
manager.notify(notification.sessionId, ALERT_ID, built)
}
/**
* The type Android 14+ requires a foreground service to declare, and nothing before it.
*
* Named behind a version check rather than passed as a constant: the value is inlined at
* compile time and would be handed to platforms that have no concept of it, which is exactly
* the case lint's InlinedApi exists to catch. Zero is what ServiceCompat wants where types do
* not apply.
*/
private fun foregroundType(): Int =
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.UPSIDE_DOWN_CAKE) {
ServiceInfo.FOREGROUND_SERVICE_TYPE_SPECIAL_USE
} else {
0
}
private fun ongoingNotification(): Notification =
NotificationCompat.Builder(this, ONGOING_CHANNEL)
.setContentTitle("Watching for sessions that need you")
.setSmallIcon(android.R.drawable.stat_notify_sync)
.setOngoing(true)
.setPriority(NotificationCompat.PRIORITY_MIN)
.build()
companion object {
/**
* Starts the service if there is a server to connect to, and stops it otherwise.
*
* Called on every launch rather than once: a service Android killed does not restart itself
* if the process was replaced, and asking for one that is already running is free.
*/
fun sync(context: Context) {
val intent = Intent(context, NotificationService::class.java)
if (loadServerSettings(context) == null) {
context.stopService(intent)
return
}
createChannels(context)
ContextCompat.startForegroundService(context, intent)
}
/**
* Two channels, because they are two different things to be told.
*
* The alerts are what somebody turned this on for, so they get the default importance and
* whatever sound and heads-up display the person has chosen for the app. The ongoing one is
* the platform's tax for staying connected, so it takes the lowest importance that exists.
* Both are created before the service starts, since posting to a channel that does not
* exist is silently dropped.
*/
private fun createChannels(context: Context) {
val manager = NotificationManagerCompat.from(context)
manager.createNotificationChannel(
NotificationChannelCompat.Builder(
ALERT_CHANNEL,
NotificationManagerCompat.IMPORTANCE_DEFAULT,
)
.setName("Sessions needing attention")
.build()
)
manager.createNotificationChannel(
NotificationChannelCompat.Builder(
ONGOING_CHANNEL,
NotificationManagerCompat.IMPORTANCE_MIN,
)
.setName("Staying connected")
.build()
)
}
private const val ALERT_CHANNEL = "sessions"
private const val ONGOING_CHANNEL = "connection"
private const val ONGOING_ID = 1
/** Shared by every alert; the session id is the tag that separates them. */
private const val ALERT_ID = 2
private const val RECONNECT_DELAY_MS = 5_000L
}
}
/** One frame of `GET /notifications`. */
data class SessionNotification(
val sessionId: String,
val title: String,
/** The wire's word: "awaitingInput" or "finished". */
val kind: String,
val at: Double,
)
fun parseNotification(json: String): SessionNotification {
val body = JSONObject(json)
return SessionNotification(
sessionId = body.getString("sessionId"),
title = body.getString("title"),
kind = body.getString("kind"),
at = body.optDouble("at", 0.0),
)
}
app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt
+4 -3
View File
@@ -263,11 +263,12 @@ private fun SessionCard(
Spacer(Modifier.height(4.dp))
Row(modifier = Modifier.fillMaxWidth()) {
Text(
// Provider, then where it runs -- "on <host>" rather
// than a bare name, so a host isn't mistaken for a model.
// Machine, then what runs on it, then what it is set to: the same order
// and separator as the session screen's header and the usage dialog, so
// one pair of facts is not written three ways.
listOfNotNull(
session.setupName,
session.provider,
"on ${session.setupName}",
session.model?.let { modelLabel(it) },
)
.joinToString(" · "),
app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt
+37 -16
View File
@@ -410,7 +410,10 @@ fun SessionScreen(
val scope = rememberCoroutineScope()
var items by remember { mutableStateOf(listOf<TranscriptItem>()) }
var status by remember { mutableStateOf(summary.status) }
var totalTokens by remember { mutableLongStateOf(0L) }
// Seeded from the row this screen was opened from, so a conversation that has spent
// something says so before any turn happens here. Zero used to mean "nothing yet" and
// "nothing in the page I loaded" at once, and the second one is most of the long sessions.
var totalTokens by remember(summary.id) { mutableLongStateOf(summary.totalTokens) }
// When this screen saw the current compaction start, on this device's own clock, and how long
// ago that is. See `compactingLabel`: null is the honest answer whenever the start was not
// witnessed here, which is what opening a session that is already compacting looks like.
@@ -495,7 +498,13 @@ fun SessionScreen(
moreHistory = entry.seq > 1L
}
when (val event = entry.event) {
is SessionEvent.UsageDelta -> totalTokens += event.tokens
// Taken, not accumulated: the server's running total is on the event, and adding
// up the deltas this screen happened to receive counted one page of a conversation
// and called it the whole. `max` because pages arrive in no guaranteed order and an
// older event's total is a smaller true answer, never a correction downwards; it
// also leaves the seeded figure alone for transcripts recorded before the backend
// sent a total at all.
is SessionEvent.UsageDelta -> totalTokens = maxOf(totalTokens, event.total)
else -> {
// What the session says it is set to now, which is the only thing that
// says it: picking from either menu asks, and the answer comes back here.
@@ -839,15 +848,18 @@ fun SessionScreen(
Spacer(Modifier.width(8.dp))
Column(Modifier.weight(1f)) {
Text(title, style = MaterialTheme.typography.titleMedium)
// Machine first, then what runs on it -- the same order and the same wording
// everywhere this pair appears, so it reads as one fact rather than as two
// sentences with different grammar. The "on" that used to sit in the middle
// made it a phrase, which only works in one order and stops working the moment
// the pair is shown anywhere else.
//
// No model. The picker in the footer already shows what this session is set to,
// and showing it twice means two things to keep in step -- they disagreed for a
// moment on every model change, since one follows the request and the other the
// session's own answer.
Text(
listOfNotNull(
summary.provider,
"on ${summary.setupName}",
// What the session says it is set to now, which is the same fact
// the picker below shows and has to be the same answer.
model?.let { modelLabel(it) },
)
.joinToString(" · "),
"${summary.setupName} · ${summary.provider}",
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
@@ -1344,17 +1356,26 @@ private fun SessionStatusRow(
)
Spacer(Modifier.weight(1f))
}
else -> {
if (status == "exited") {
// Every remaining state says which one it is, including the quiet one. The row used
// to name only `exited` and leave the rest blank, so a session sitting idle and one
// whose status nobody could read looked identical -- and a turn that had just been
// stopped showed nothing at all, which reads as the app having lost the session
// rather than as the stop having worked. The words are the session list's own, so
// one state is not called two things depending which screen you are on.
else ->
Text(
"exited",
when (status) {
"idle" -> "idle"
"exited" -> "exited"
"awaitingInput" -> "your turn"
"unknown" -> "can't tell"
else -> status
},
style = MaterialTheme.typography.labelSmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
modifier = Modifier.weight(1f),
)
}
Spacer(Modifier.weight(1f))
}
}
// Nothing rather than "0 tok" before anything has been spent: a total of zero is a fact
// about a conversation that has not started, and it is the one reading nobody needs.
if (totalTokens > 0) {
app/androidApp/src/main/kotlin/com/example/aiapp/SessionSettingsScreen.kt
+73 -1
View File
@@ -15,8 +15,10 @@ import androidx.compose.foundation.verticalScroll
import androidx.compose.material3.Button
import androidx.compose.material3.MaterialTheme
import androidx.compose.material3.OutlinedTextField
import androidx.compose.material3.Switch
import androidx.compose.material3.Text
import androidx.compose.runtime.Composable
import androidx.compose.runtime.LaunchedEffect
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
@@ -52,6 +54,43 @@ fun SessionSettingsScreen(
var name by remember(session.id) { mutableStateOf(session.title) }
var saving by remember { mutableStateOf(false) }
var error by remember { mutableStateOf<String?>(null) }
// Null until the server has been asked. The row this screen was opened from is a snapshot of
// whenever the list was last fetched, so drawing the switch straight from it would show a
// position that may have been changed since -- from here or from another device -- with
// nothing to say so. Until the answer arrives the switch is disabled and the caption says it
// is being read, which is the one honest thing a two-position control can do about not
// knowing.
var notify by remember(session.id) { mutableStateOf<Boolean?>(null) }
var notifyError by remember { mutableStateOf<String?>(null) }
LaunchedEffect(session.id) {
notify =
try {
withContext(Dispatchers.IO) { fetchSession(settings, session.id).notify }
} catch (e: ApiException) {
// Left unknown rather than falling back to the stale row: the switch stays
// disabled and says why, instead of offering a position nothing confirmed.
notifyError = e.message
null
}
}
// Moved optimistically so the switch answers the finger that moved it, and put back if the
// request is refused -- a switch that waits for a round trip reads as broken on a slow
// tunnel, and one that stays moved after a refusal lies.
fun setNotify(wanted: Boolean) {
val was = notify
notify = wanted
notifyError = null
scope.launch {
try {
withContext(Dispatchers.IO) { setSessionNotify(settings, session.id, wanted) }
} catch (e: ApiException) {
notify = was
notifyError = 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.
@@ -106,7 +145,40 @@ fun SessionSettingsScreen(
color = MaterialTheme.colorScheme.onSurfaceVariant,
modifier = Modifier.padding(top = 4.dp),
)
Spacer(Modifier.height(16.dp))
Spacer(Modifier.height(24.dp))
// The switch gets its own row rather than sitting beside the label: a control is taller
// than a line of text, so putting one in a row with a label re-centres that label and
// knocks it out of line with everything above it.
Row(verticalAlignment = Alignment.CenterVertically, modifier = Modifier.fillMaxWidth()) {
Text("Notify me", modifier = Modifier.weight(1f))
Switch(
checked = notify == true,
onCheckedChange = { setNotify(it) },
enabled = notify != null,
)
}
Text(
if (notify == null && notifyError == null) "Reading the current setting..."
else
"A notification when this session asks you something or finishes a turn. Kept " +
"on the backend, so every device agrees about which sessions are worth " +
"interrupting you for.",
style = MaterialTheme.typography.bodySmall,
color = MaterialTheme.colorScheme.onSurfaceVariant,
)
notifyError?.let {
Spacer(Modifier.height(4.dp))
// Beside the switch that failed, not with the rename's error: they are two requests
// and a reader has to be able to tell which one the server refused.
Text(
it,
color = MaterialTheme.colorScheme.error,
style = MaterialTheme.typography.bodySmall,
)
}
Spacer(Modifier.height(24.dp))
// Disabled rather than absent while there is nothing to save: a button that comes and
// goes makes its own presence the signal, and its absence cannot say why.
Button(onClick = { save() }, enabled = changed && !saving) {
app/androidApp/src/main/kotlin/com/example/aiapp/Theme.kt
+1 -10
View File
@@ -167,16 +167,7 @@ fun quotaColor(percent: Double): Color =
private const val WARNING_PERCENT = 75.0
/** Close enough that the next turn may be the one that is refused. */
private const val OVER_LIMIT_PERCENT = 95.0
/**
* Code: a fenced block, an inline span, a tool's input.
*
* Green because on this palette it is what a literal is coloured as, and because code sits on
* Surface 0 where the ordinary text colour would say nothing about it being code.
*/
val codeColor: Color
@Composable get() = Mocha.Green
private const val OVER_LIMIT_PERCENT = 90.0
/**
* Catppuccin Mocha as a syntax theme, for the highlighter used on a tool call's input.
app/run-android.sh
+10 -3
View File
@@ -9,15 +9,22 @@ set -eu
APP_ID="com.example.aiapp"
# The AVD shared by this machine's Android projects -- one emulator, not
# one per repo. Override with AVD_NAME=... elsewhere.
AVD_NAME="${AVD_NAME:-tdep}"
DEVICE_PROFILE="${DEVICE_PROFILE:-pixel_10}"
SYSTEM_IMAGE="${SYSTEM_IMAGE:-system-images;android-36;google_apis;x86_64}"
SCRIPT_DIR=$(cd "$(dirname "$0")" && pwd)
cd "$SCRIPT_DIR"
# One AVD per checkout, named after it -- so two clones of this repo, or a
# clone and a worktree, each get their own rather than fighting over one.
# This used to default to a machine-wide "tdep", which made the emulator the
# one thing here that could not be worked on in parallel: taking it meant
# asking whoever had it, waiting, and handing it back, and installing onto a
# running one steals the foreground from whatever they were looking at.
# Derived rather than written down, so neither clone names the other's.
# Override with AVD_NAME=... to share one deliberately.
AVD_NAME="${AVD_NAME:-$(basename "$(dirname "$SCRIPT_DIR")")}"
# shellcheck source=./android-env.sh
. ./android-env.sh
server/Cargo.lock
Generated
+1
View File
@@ -1647,6 +1647,7 @@ dependencies = [
"futures-core",
"pin-project-lite",
"tokio",
"tokio-util",
]
[[package]]
server/Cargo.toml
+4 -1
View File
@@ -16,7 +16,10 @@ wg-app-link = { path = "../wg-app-link/server" }
axum = { version = "0.8", features = ["json", "multipart"] }
axum-server = { version = "0.8", features = ["tls-rustls"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time", "process", "io-util", "signal"] }
tokio-stream = "0.1"
# `sync` for BroadcastStream: the notifications route turns the manager's
# broadcast channel straight into an SSE body, which is the one place here a
# broadcast receiver has to be a Stream rather than something to poll.
tokio-stream = { version = "0.1", features = ["sync"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
serde = { version = "1", features = ["derive"] }
server/src/config.rs
+19
View File
@@ -206,10 +206,28 @@ pub struct SessionConfig {
/// across writes.
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub params: BTreeMap<String, String>,
/// Whether a phone should be told when this session wants attention.
///
/// Stored here rather than on the phone because it is a fact about the
/// session: one that runs unattended overnight should be quiet on
/// every device, and answering that question again on each new phone
/// is how two devices come to disagree about which sessions matter.
///
/// Defaults to on, and on for a config written before this field
/// existed. The alternative -- silent unless asked -- makes the
/// feature invisible to anyone who does not go looking for it, and a
/// notification nobody wanted is turned off in one tap where one that
/// never arrived is not diagnosable at all.
#[serde(default = "notify_default")]
pub notify: bool,
/// Epoch seconds when the session was spawned.
pub created: f64,
}
fn notify_default() -> bool {
true
}
/// The name of the echo provider, and of the setup this machine gets on
/// first run.
///
@@ -376,6 +394,7 @@ mod tests {
cwd: None,
permission_mode: None,
params: BTreeMap::new(),
notify: true,
created: 1234.5,
}],
};
server/src/routes.rs
+69 -2
View File
@@ -10,6 +10,7 @@
//! PUT /setups/{id} rename {name?} and/or re-probe {rediscover?}
//! DELETE /setups/{id} remove, refused while sessions use it
//! GET /sessions list (id, provider, title, model, status, last activity)
//! GET /sessions/{id} one session, for refetching after a change
//! POST /sessions spawn {setup, provider, title?, model?, cwd?, params?}
//! GET /sessions/{id}/events?after=N SSE: backlog after N, then live
//! (a backlog past CATCH_UP_LIMIT arrives as a
@@ -24,6 +25,9 @@
//! POST /sessions/{id}/attachments multipart image upload -> {id}, referenced by /message
//! GET /sessions/{id}/files/{name} images the session produced or was sent
//! DELETE /sessions/{id} kill process, delete transcript + files
//! POST /sessions/{id}/notify {notify} -- announce this one or not
//! GET /notifications SSE: every session's attention-wanting
//! moments, live only (see `notifications`)
//! GET /usage cached usage windows per provider
//! ```
//!
@@ -57,7 +61,7 @@ use axum::routing::{delete, get, post};
use serde::Deserialize;
use tokio::sync::{broadcast, mpsc};
use tokio_stream::StreamExt;
use tokio_stream::wrappers::ReceiverStream;
use tokio_stream::wrappers::{BroadcastStream, ReceiverStream};
use crate::session::driver::SessionCommand;
use crate::session::transcript::{CATCH_UP_LIMIT, CatchUp, SeqEvent, catch_up};
@@ -77,7 +81,7 @@ pub fn router(manager: Arc<SessionManager>) -> Router {
get(read_setup).put(update_setup).delete(delete_setup),
)
.route("/sessions", get(list_sessions).post(spawn_session))
.route("/sessions/{id}", delete(delete_session))
.route("/sessions/{id}", get(read_session).delete(delete_session))
.route("/sessions/{id}/events", get(events))
.route("/sessions/{id}/transcript", get(transcript))
.route("/sessions/{id}/message", post(message))
@@ -86,6 +90,8 @@ pub fn router(manager: Arc<SessionManager>) -> Router {
.route("/sessions/{id}/title", post(rename))
.route("/sessions/{id}/model", post(set_model))
.route("/sessions/{id}/permission-mode", post(set_permission_mode))
.route("/sessions/{id}/notify", post(set_notify))
.route("/notifications", get(notifications))
.route("/sessions/{id}/compact", post(compact))
.route("/sessions/{id}/command", post(command))
.route("/sessions/{id}/attachments", post(upload_attachment))
@@ -144,6 +150,26 @@ async fn list_sessions(State(manager): State<Arc<SessionManager>>) -> axum::Json
axum::Json(manager.sessions())
}
/// One session's row, for a screen that has to show what is true now.
///
/// The list is a snapshot taken when somebody last looked at it, and a
/// screen opened from a row carries that snapshot with it. That is fine for
/// what a row *says* and wrong for what a control is *set to*: a switch
/// drawn from a stale row shows the position it had when the list was
/// fetched, which may be minutes and another device ago, and the person
/// reading it cannot tell. Same reason `GET /setups/{id}` exists.
async fn read_session(
State(manager): State<Arc<SessionManager>>,
UrlPath(id): UrlPath<String>,
) -> Result<axum::Json<SessionInfo>, ApiError> {
manager
.sessions()
.into_iter()
.find(|session| session.id == id)
.map(axum::Json)
.ok_or_else(|| ApiError::NotFound(format!("no session {id}")))
}
/// What the spawn screen needs to render itself, so the phone holds no
/// hardcoded list: a setup added to `config.ron` shows up with no app
/// rebuild.
@@ -740,6 +766,23 @@ async fn set_permission_mode(
Ok(StatusCode::NO_CONTENT)
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct NotifyRequest {
notify: bool,
}
async fn set_notify(
State(manager): State<Arc<SessionManager>>,
UrlPath(id): UrlPath<String>,
axum::Json(body): axum::Json<NotifyRequest>,
) -> Result<StatusCode, ApiError> {
manager
.set_session_notify(&id, body.notify)
.map_err(bad_request)?;
Ok(StatusCode::NO_CONTENT)
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct CommandRequest {
@@ -910,6 +953,30 @@ async fn events(
Ok(Sse::new(ReceiverStream::new(stream).map(Ok)).keep_alive(KeepAlive::default()))
}
/// Every session's attention-wanting moments, on one stream.
///
/// **Live only, with no cursor**, which is the one place this server does not
/// offer to catch a client up. A notification is a claim about now: replaying
/// "your turn" from an hour ago tells somebody to go and look at a session
/// that may have been answered from another device since, and a notification
/// that is wrong is worse than one that never came -- it costs the reader the
/// trip *and* teaches them to distrust the next one. What was missed while
/// disconnected is still on the session list, which is the surface that
/// answers "what is waiting" without claiming to be news.
async fn notifications(
State(manager): State<Arc<SessionManager>>,
) -> Sse<impl tokio_stream::Stream<Item = Result<SseEvent, Infallible>>> {
let live = manager.subscribe_notifications();
let stream = BroadcastStream::new(live).filter_map(|item| {
// A lagged subscriber has lost the oldest notifications, and there is
// nothing useful to say about that: the ones it still gets are the
// recent ones, which are the ones worth acting on.
let notification = item.ok()?;
Some(Ok(SseEvent::default().json_data(&notification).ok()?))
});
Sse::new(stream).keep_alive(KeepAlive::default())
}
/// Feeds one SSE subscriber: transcript replay after the cursor, then live
/// events, catching back up from the file whenever the broadcast channel
/// laps us. Ends when the client disconnects (send fails) or the session
server/src/session/claude.rs
+4
View File
@@ -567,6 +567,10 @@ impl Driver for ClaudeDriver {
/// typed deliberately, and dropping it would lose a message that never
/// reached the transcript, with nothing on screen to say so.
fn interrupt(&self) {
// Recorded before the request goes out, so the result it produces is
// read as the stop somebody asked for rather than as a failure --
// see `Translator::interrupting`.
self.state.lock().unwrap().expect_interrupt();
self.send_control(json!({"subtype": "interrupt"}), None);
}
server/src/session/claude/translate.rs
+74 -3
View File
@@ -82,6 +82,20 @@ pub(super) struct Translator {
/// out is the response: every entry is removed when one arrives,
/// whether it succeeded or failed.
asked: HashMap<String, Setting>,
/// Whether this side asked the turn to stop.
///
/// The CLI reports an interrupted turn the same way it reports one that
/// broke -- a `result` with `is_error` set -- so the line itself cannot
/// tell them apart, and a person who pressed Stop was shown "the turn
/// ended with an error" for doing exactly what the button says. What
/// separates them is not in the message at all: it is that *we* asked.
/// So the driver says so before the request goes out, the same way it
/// does for a setting, and this remembers it until the result lands.
///
/// Its path out is that result -- set by `expect_interrupt`, cleared by
/// the next `result` whichever way it went, so a genuine failure in a
/// later turn is still reported.
interrupting: bool,
session_dir: PathBuf,
}
@@ -91,6 +105,7 @@ impl Translator {
session_id: None,
pending: HashMap::new(),
asked: HashMap::new(),
interrupting: false,
session_dir,
}
}
@@ -103,6 +118,13 @@ impl Translator {
pub(super) fn expect_setting(&mut self, request_id: String, setting: Setting) {
self.asked.insert(request_id, setting);
}
/// Says that the turn about to end was stopped on purpose -- see
/// [`Translator::interrupting`]. Called before the request goes out,
/// for the reason [`Translator::expect_setting`] gives.
pub(super) fn expect_interrupt(&mut self) {
self.interrupting = true;
}
pub(super) fn translate(&mut self, message: &Value) -> Vec<Event> {
// Events from subagents (Task tool internals) carry a
// parent_tool_use_id; the transcript shows the Task tool's own
@@ -182,7 +204,11 @@ impl Translator {
.and_then(Value::as_u64)
.unwrap_or(0);
let mut events = Vec::new();
if message
// Whichever way this result went, the interrupt it may have
// been answering is now spent.
let asked_to_stop = std::mem::take(&mut self.interrupting);
if !asked_to_stop
&& message
.get("is_error")
.and_then(Value::as_bool)
.unwrap_or(false)
@@ -196,7 +222,7 @@ impl Translator {
});
}
if tokens > 0 {
events.push(Event::UsageDelta { tokens });
events.push(Event::UsageDelta { tokens, total: 0 });
}
events.push(Event::Status {
state: SessionStatus::Idle,
@@ -1026,7 +1052,10 @@ mod tests {
assert_eq!(
events,
vec![
Event::UsageDelta { tokens: 182 },
Event::UsageDelta {
tokens: 182,
total: 0
},
Event::Status {
state: SessionStatus::Idle
},
@@ -1138,6 +1167,48 @@ mod tests {
);
}
/// 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 -- `is_error` set, on a `result` -- so somebody who pressed the
/// button was shown "the turn ended with an error" for doing what the
/// button says. What separates the two is not in the line: it is that
/// this side asked. The second half of this test is the one that
/// matters, because the naive fix -- never reporting an error result --
/// passes the first half and silences every genuine failure afterwards.
#[test]
fn a_turn_stopped_on_purpose_is_not_an_error() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let stopped_result = r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Interrupted by user","usage":{}}"#;
translator.expect_interrupt();
let events = translate_lines(&mut translator, &[stopped_result]);
assert!(
!events
.iter()
.any(|event| matches!(event, Event::Error { .. })),
"a stop the driver asked for was reported as a failure: {events:?}"
);
assert_eq!(
*events.last().unwrap(),
Event::Status {
state: SessionStatus::Idle
},
"an interrupted turn still has to end the turn"
);
// The interrupt is spent, so the next failure is a failure again.
let later = translate_lines(&mut translator, &[stopped_result]);
assert!(
later
.iter()
.any(|event| matches!(event, Event::Error { .. })),
"a later failure was swallowed by an interrupt that had already been answered"
);
}
#[test]
fn replayed_and_synthetic_user_text_is_skipped() {
let dir = tempfile::tempdir().expect("tempdir");
server/src/session/driver.rs
+16
View File
@@ -230,6 +230,22 @@ pub enum Event {
/// Per-turn token counts, where the dialect reports them.
UsageDelta {
tokens: u64,
/// Every token this session has spent, this turn included.
///
/// Filled in by the pump, not by drivers: a driver reports what its
/// turn cost, and only the pump sees all of them. Carried on the
/// event rather than left to be added up by whoever is reading,
/// because a reader has only *part* of the transcript -- a phone
/// opens a session on the newest page -- so a total it summed
/// itself would be the newest page's total wearing the whole
/// conversation's label. Worse when the page has no turn in it at
/// all: the count then reads zero, and a zero is drawn as nothing.
///
/// Zero on entries written before this existed, which is why the
/// pump seeds its running total by adding up `tokens` at startup
/// rather than reading the last of these.
#[serde(default)]
total: u64,
},
/// A compaction that finished, and how much context it recovered.
///
server/src/session/echo.rs
+1
View File
@@ -446,6 +446,7 @@ impl EchoDriver {
}
send(Event::UsageDelta {
tokens: text.split_whitespace().count() as u64,
total: 0,
});
finish();
});
server/src/session/llama.rs
+5 -2
View File
@@ -608,7 +608,7 @@ fn generate(
}
}
if tokens > 0 {
let _ = sink.send(Event::UsageDelta { tokens });
let _ = sink.send(Event::UsageDelta { tokens, total: 0 });
}
Ok(())
}
@@ -712,7 +712,10 @@ mod tests {
Event::AssistantText {
delta: "still here".into(),
},
Event::UsageDelta { tokens: 12 },
Event::UsageDelta {
tokens: 12,
total: 12,
},
]);
let messages = conversation(&path);
assert_eq!(messages.len(), 2);
server/src/session/mod.rs
+376 -2
View File
@@ -42,6 +42,14 @@ use transport::Transport;
/// the size only bounds memory, not correctness.
const EVENT_BUFFER: usize = 256;
/// Fan-out buffer for notifications, across every session.
///
/// Small, and deliberately: a subscriber that falls this far behind on a
/// stream carrying two events per turn is not one whose backlog is worth
/// delivering. Lagging drops the oldest, which is the right end to lose --
/// the newest "your turn" is the one still true.
const NOTIFICATION_BUFFER: usize = 64;
pub fn now() -> f64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
@@ -62,6 +70,37 @@ pub struct SpawnSpec {
pub params: std::collections::BTreeMap<String, String>,
}
/// A moment worth interrupting somebody for, as `GET /notifications`
/// sends it.
///
/// Two kinds, and the pair is the whole feature: a session that has *asked*
/// something cannot continue until it is answered, and one that has
/// *finished* is work somebody walked away from. Everything else a session
/// does is progress they did not ask to be told about.
///
/// Carries the title rather than only the id, so the phone can write the
/// notification without a round trip -- it may well be showing no screen at
/// all when this arrives.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Notification {
pub session_id: String,
pub title: String,
pub kind: NotificationKind,
/// Epoch seconds, so a phone that was asleep can say how long ago.
pub at: f64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum NotificationKind {
/// The session is waiting on a person: a question, or a permission.
AwaitingInput,
/// A turn ended without one. Only ever sent for a session that was
/// *seen* running -- see `notification_for`.
Finished,
}
/// One row of `GET /sessions`.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
@@ -104,6 +143,13 @@ pub struct SessionInfo {
pub imported: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub cwd: Option<PathBuf>,
/// Every token this session has spent, so a phone showing a total does
/// not have to add up a transcript it only holds part of.
pub total_tokens: u64,
/// Whether this session announces itself -- reported for the same
/// 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,
pub status: SessionStatus,
pub last_activity: f64,
pub created: f64,
@@ -210,6 +256,20 @@ struct Shared {
/// session was *launched* with, so reporting from it would show the
/// mode a change had already replaced.
permission_mode: Mutex<Option<String>>,
/// Every token this session has spent.
///
/// Kept here because only the pump sees every turn, and reported on the
/// session row so a phone opening a long conversation has the real
/// figure rather than the newest page's share of it.
total_tokens: Mutex<u64>,
/// Whether this session's attention-wanting moments are announced.
///
/// Mirrored out of the config so the pump can read it without taking
/// the manager's lock -- the pump runs underneath the manager and
/// reaching back up for a field would invert that. `set_session_notify`
/// writes both, in that order, which is the same shape every other
/// live-and-persisted setting here uses.
notify: Mutex<bool>,
/// How many events this session has ever recorded.
///
/// Only the import sync reads it, and only to answer one question:
@@ -319,6 +379,8 @@ impl LiveSession {
title: self.shared.title.lock().unwrap().clone(),
model: self.shared.model.lock().unwrap().clone(),
permission_mode: self.shared.permission_mode.lock().unwrap().clone(),
total_tokens: *self.shared.total_tokens.lock().unwrap(),
notify: *self.shared.notify.lock().unwrap(),
imported,
keeps_own_transcript,
cwd: self.meta.cwd.clone(),
@@ -343,6 +405,10 @@ pub struct SessionManager {
/// which is why they live beside the session directories rather than
/// inside one.
models_dir: PathBuf,
/// Where every session's pump sends what a phone should be told about.
/// Held here rather than per session for the reason
/// [`SessionManager::subscribe_notifications`] gives.
notifications: broadcast::Sender<Notification>,
inner: RwLock<Inner>,
}
@@ -356,6 +422,7 @@ impl SessionManager {
let config = Config::load(&config_path)?;
wg_app_link::private::create_dir(&data_dir)?;
let (notifications, _) = broadcast::channel(NOTIFICATION_BUFFER);
let mut live = HashMap::new();
for meta in &config.sessions {
// One unlaunchable session -- a corrupt transcript, an
@@ -370,6 +437,7 @@ impl SessionManager {
&data_dir,
&models_dir,
None,
notifications.clone(),
)
}) {
Ok(session) => {
@@ -384,6 +452,7 @@ impl SessionManager {
config_path,
data_dir,
models_dir,
notifications,
inner: RwLock::new(Inner { config, live }),
};
Ok(manager)
@@ -647,6 +716,8 @@ impl SessionManager {
title: meta.title.clone(),
model: meta.model.clone(),
permission_mode: meta.permission_mode.clone(),
total_tokens: 0,
notify: meta.notify,
imported: import::read_cursor(&self.data_dir.join(&meta.id)).is_some(),
keeps_own_transcript: keeps_own_transcript(
&inner.config,
@@ -662,6 +733,16 @@ impl SessionManager {
.collect()
}
/// Every session's attention-wanting moments, on one stream.
///
/// One connection for the whole backend rather than one per session:
/// the phone subscribes to this while showing no session at all, and a
/// connection per session would mean opening one for every session that
/// exists in order to hear about any of them.
pub fn subscribe_notifications(&self) -> broadcast::Receiver<Notification> {
self.notifications.subscribe()
}
pub fn session(&self, id: &str) -> Option<Arc<LiveSession>> {
self.inner.read().unwrap().live.get(id).cloned()
}
@@ -739,6 +820,11 @@ impl SessionManager {
cwd: spec.cwd,
permission_mode: spec.permission_mode,
params: spec.params,
// On by default -- see `SessionConfig::notify`. Not offered at
// spawn: a session's first turn is exactly the one somebody is
// waiting for, and a switch on the spawn screen would be a
// decision asked before there is anything to decide about.
notify: true,
created: now(),
};
@@ -749,6 +835,7 @@ impl SessionManager {
&self.data_dir,
&self.models_dir,
seed,
self.notifications.clone(),
)?;
let mut candidate = inner.config.clone();
candidate.sessions.push(meta);
@@ -804,6 +891,33 @@ impl SessionManager {
Ok(())
}
/// Turns this session's notifications on or off, live and persisted.
///
/// Both, in that order, for the reason every setting here writes both:
/// the config decides what a restart believes and the live copy decides
/// what the running pump does, and a change that lands in one of them is
/// a switch that moves back on its own.
///
/// Nothing is told to the driver. Unlike the model or the permission
/// mode, this changes nothing about how the session runs -- it is about
/// who gets told, and the session is not the one being told.
pub fn set_session_notify(&self, id: &str, notify: bool) -> 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.notify = notify;
}
candidate.save(&self.config_path)?;
inner.config = candidate;
if let Some(session) = inner.live.get(id) {
*session.shared.notify.lock().unwrap() = notify;
}
Ok(())
}
/// Renames a session: persisted, shown, and passed on to whatever is
/// running it.
///
@@ -1084,6 +1198,7 @@ fn launch(
data_dir: &Path,
models_dir: &Path,
seed: Option<Seed>,
notifications: broadcast::Sender<Notification>,
) -> Result<Arc<LiveSession>> {
let dir = data_dir.join(&meta.id);
wg_app_link::private::create_dir(&dir)?;
@@ -1109,9 +1224,15 @@ fn launch(
// this is then the only true answer available.
status: Mutex::new(transcript.last_status().unwrap_or(SessionStatus::Idle)),
title: Mutex::new(meta.title.clone()),
last_activity: Mutex::new(now()),
// What the transcript last recorded, not the clock: this server has
// just been told nothing, and `now()` claimed every relaunched
// session had been active this instant -- see
// `Transcript::last_activity`.
last_activity: Mutex::new(transcript.last_activity().unwrap_or_else(now)),
model: Mutex::new(meta.model.clone()),
permission_mode: Mutex::new(meta.permission_mode.clone()),
total_tokens: Mutex::new(transcript.total_tokens()),
notify: Mutex::new(meta.notify),
written: Mutex::new(0),
});
@@ -1156,11 +1277,13 @@ fn launch(
});
tokio::spawn(pump(
meta.id.clone(),
transcript,
source,
Arc::clone(&shared),
events.clone(),
Arc::clone(&commands),
notifications,
));
Ok(Arc::new(LiveSession {
@@ -1205,12 +1328,33 @@ fn is_news(event: &Event, shared: &Shared) -> bool {
}
}
/// Whether moving from `was` to `now` is worth interrupting somebody for.
///
/// The asymmetry is the point. *Waiting on a person* is worth saying however
/// the session got there -- it is a question that will sit unanswered until
/// somebody sees it. *Finished* is only worth saying when this server
/// watched the work happen: a session settling into idle because it was
/// adopted at startup, or because a driver announced itself, is not news
/// that anything ended, and sending it would put "finished" on the phone for
/// every session in the config every time the backend restarts.
fn notification_for(was: SessionStatus, now: SessionStatus) -> Option<NotificationKind> {
match (was, now) {
(_, SessionStatus::AwaitingInput) => Some(NotificationKind::AwaitingInput),
(SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle) => {
Some(NotificationKind::Finished)
}
_ => None,
}
}
async fn pump(
id: String,
mut transcript: Transcript,
mut source: mpsc::UnboundedReceiver<Event>,
shared: Arc<Shared>,
events: broadcast::Sender<SeqEvent>,
commands: Arc<Commands>,
notifications: broadcast::Sender<Notification>,
) {
while let Some(event) = source.recv().await {
let ts = now();
@@ -1220,6 +1364,18 @@ async fn pump(
// for where a user's message sits: where the session read it.
let event = match event {
Event::MessageTaken { id, text } => Event::UserMessage { id, text },
// The running total is the pump's to keep, for the reason the
// field gives: a driver knows what its own turn cost and
// nothing else does. Added here rather than at each driver so
// a new one cannot get it wrong by leaving it out.
Event::UsageDelta { tokens, .. } => {
let mut total = shared.total_tokens.lock().unwrap();
*total += tokens;
Event::UsageDelta {
tokens,
total: *total,
}
}
other => other,
};
// Nothing changed, so there is nothing to record. Both of these
@@ -1251,7 +1407,21 @@ async fn pump(
match transcript.append(event, ts) {
Ok(entry) => {
if let Event::Status { state } = &entry.event {
*shared.status.lock().unwrap() = *state;
// Read before it is overwritten: what makes a status
// worth announcing is the transition, not the value.
let was = std::mem::replace(&mut *shared.status.lock().unwrap(), *state);
if let Some(kind) =
notification_for(was, *state).filter(|_| *shared.notify.lock().unwrap())
{
// No subscribers is the ordinary case -- nobody has
// the app open -- and it is not an error.
let _ = notifications.send(Notification {
session_id: id.clone(),
title: shared.title.lock().unwrap().clone(),
kind,
at: ts,
});
}
}
*shared.last_activity.lock().unwrap() = ts;
*shared.written.lock().unwrap() += 1;
@@ -1366,6 +1536,101 @@ mod tests {
assert!(!DriverKind::LlamaCpp.keeps_own_transcript());
}
/// The two transitions worth interrupting somebody for, and the ones
/// that look like them and are not.
///
/// The idle cases are the whole reason this is a function rather than a
/// pair of `if`s at the callsite. A session settles into idle for
/// several reasons that are not "your work finished": it was adopted at
/// startup, its driver announced itself, it came back from a state
/// nobody could read. Announcing those would put "finished" on the phone
/// for every session in the config every time the backend restarts,
/// which is the failure that makes somebody turn the whole feature off.
#[test]
fn only_a_watched_turn_ending_counts_as_finished() {
use NotificationKind::{AwaitingInput, Finished};
use SessionStatus::{Compacting, Exited, Idle, Running, Unknown};
// Waiting on a person is worth saying however it was reached: it
// will sit unanswered until somebody is told.
assert_eq!(
notification_for(Running, SessionStatus::AwaitingInput),
Some(AwaitingInput)
);
assert_eq!(
notification_for(Idle, SessionStatus::AwaitingInput),
Some(AwaitingInput)
);
// A turn this server watched run, ending.
assert_eq!(notification_for(Running, Idle), Some(Finished));
assert_eq!(notification_for(Compacting, Idle), Some(Finished));
// Idle arrived at from anywhere else is not an ending.
assert_eq!(notification_for(Idle, Idle), None);
assert_eq!(notification_for(Unknown, Idle), None);
assert_eq!(notification_for(Exited, Idle), None);
assert_eq!(notification_for(SessionStatus::AwaitingInput, Idle), None);
// Everything else a session does is progress nobody asked to hear.
assert_eq!(notification_for(Idle, Running), None);
assert_eq!(notification_for(Running, Compacting), None);
assert_eq!(notification_for(Running, Exited), None);
}
/// The switch reaches the running pump, not just the config file.
///
/// The failure this exists for is silent in the direction that matters:
/// a `set_session_notify(false)` that wrote only the config would look
/// correct on the settings screen and in the file, and keep notifying
/// until the backend was restarted. Nothing on screen would say so, and
/// the person who turned it off is by definition not watching.
#[tokio::test]
async fn turning_notifications_off_stops_them_without_a_restart() {
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");
let session = manager.session(&info.id).expect("live");
let mut notifications = manager.subscribe_notifications();
session.send_message("hello".to_string(), Vec::new());
let first = tokio::time::timeout(Duration::from_secs(5), notifications.recv())
.await
.expect("a notification within five seconds")
.expect("channel open");
assert_eq!(first.kind, NotificationKind::Finished);
assert_eq!(first.session_id, info.id);
// The title travels with it, because the phone may have no screen
// open to look one up on.
assert_eq!(first.title, session.info("m", false, false).title);
manager.set_session_notify(&info.id, false).expect("off");
// Subscribed before the message, or the turn can finish in the gap
// and leave this waiting for an event that has already gone past.
let mut events = session.subscribe();
session.send_message("hello again".to_string(), Vec::new());
// The turn still happens -- this is a switch about being told, not
// about running -- so wait for the turn's own event and then check
// that nothing was announced alongside it.
collect_until(&mut events, |event| {
matches!(
event,
Event::Status {
state: SessionStatus::Idle
}
)
})
.await;
assert!(
notifications.try_recv().is_err(),
"a session with notifications off still announced itself"
);
}
/// A session this app *spawned* is one it is driving, and used to look
/// like somebody else's.
///
@@ -1656,6 +1921,115 @@ mod tests {
)));
}
#[tokio::test]
async fn a_restart_reports_when_a_session_last_did_something() {
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.clone(),
data_dir.clone(),
data_dir.join("models"),
)
.expect("manager");
let info = manager.spawn_session(echo_spec()).expect("spawn");
let session = manager.session(&info.id).expect("live session");
let mut rx = session.subscribe();
session.send_message("something".to_string(), Vec::new());
collect_turn(&mut rx).await;
let before_restart = manager.sessions()[0].last_activity;
drop(rx);
drop(session);
drop(manager);
// Far enough back that a restart taking the clock cannot pass by
// being fast: the assertion is about which source was used, not
// about how long the test took.
let long_ago = before_restart - 86_400.0;
rewrite_transcript_times(&data_dir.join(&info.id).join("transcript.jsonl"), long_ago);
let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
.expect("manager restart");
let listed = manager.sessions();
assert_eq!(listed.len(), 1);
assert!(
(listed[0].last_activity - long_ago).abs() < 1.0,
"a relaunched session reported {} instead of the {long_ago} its transcript records \
-- every row would read \"just now\" and the list would sort by nothing",
listed[0].last_activity,
);
}
/// The total covers the whole conversation, not the part a reader holds.
///
/// The bug this fixes was invisible in exactly the way that matters: a
/// phone opens a session on its newest page and used to add up the
/// `UsageDelta`s it found there, so a long conversation reported its
/// last few turns as the total -- and a page with no turn in it at all
/// reported nothing, since zero is drawn as blank. Both readings looked
/// like an answer.
#[tokio::test]
async fn the_token_total_covers_the_whole_conversation() {
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.clone(),
data_dir.clone(),
data_dir.join("models"),
)
.expect("manager");
let info = manager.spawn_session(echo_spec()).expect("spawn");
let session = manager.session(&info.id).expect("live");
// Echo charges a token a word, so the arithmetic is checkable.
let mut rx = session.subscribe();
session.send_message("one two three".to_string(), Vec::new());
collect_turn(&mut rx).await;
session.send_message("four five".to_string(), Vec::new());
collect_turn(&mut rx).await;
let running = manager.sessions()[0].total_tokens;
assert_eq!(running, 5, "two turns of three and two words");
// The event carries it too, so a phone never has to add up its own.
let last_total = transcript::read_after(session.transcript_path(), 0)
.expect("transcript")
.iter()
.rev()
.find_map(|entry| match entry.event {
Event::UsageDelta { total, .. } => Some(total),
_ => None,
})
.expect("a usage event");
assert_eq!(last_total, running);
// And a restart picks it up from the file rather than starting over.
drop(rx);
drop(session);
drop(manager);
let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
.expect("manager restart");
assert_eq!(manager.sessions()[0].total_tokens, running);
}
/// Backdates every line in a transcript, so a restart has something to
/// report that the clock could not have produced.
fn rewrite_transcript_times(path: &Path, ts: f64) {
let text = std::fs::read_to_string(path).expect("read transcript");
let rewritten: String = text
.lines()
.map(|line| {
let mut entry: serde_json::Value = serde_json::from_str(line).expect("line");
entry["ts"] = serde_json::json!(ts);
format!("{entry}\n")
})
.collect();
std::fs::write(path, rewritten).expect("write transcript");
}
#[tokio::test]
async fn a_restart_relaunches_sessions_and_continues_the_numbering() {
let dir = tempfile::tempdir().expect("tempdir");
server/src/session/transcript.rs
+46 -3
View File
@@ -31,14 +31,16 @@ pub struct Transcript {
file: File,
next_seq: u64,
last_status: Option<SessionStatus>,
last_activity: Option<f64>,
total_tokens: u64,
}
impl Transcript {
/// Opens (or creates) the log at `path`, continuing the sequence from
/// the last line if one exists.
pub fn open(path: &Path) -> Result<Self> {
// One pass for both answers. They are wanted at the same moment by
// the same caller, and reading the file twice to get them doubled
// One pass for all three answers. They are wanted at the same moment
// by the same caller, and reading the file again for each doubled
// the cost of starting every session -- which is paid per session,
// at the point a restart is trying to be quick.
let existing = read_after(path, 0)?;
@@ -59,6 +61,17 @@ impl Transcript {
file,
next_seq: last_seq + 1,
last_status,
last_activity: existing.last().map(|entry| entry.ts),
// Added up rather than read off the newest entry: `total` is a
// later addition, so a transcript written before it has zero on
// every line while `tokens` was always there.
total_tokens: existing
.iter()
.map(|entry| match entry.event {
Event::UsageDelta { tokens, .. } => tokens,
_ => 0,
})
.sum(),
})
}
@@ -76,6 +89,33 @@ impl Transcript {
self.last_status
}
/// When this session last did anything, as of opening.
///
/// Read from the file for the same reason [`Transcript::last_status`]
/// is, and it is the same mistake in the other direction: a restarting
/// server has been told nothing, and taking the clock instead said every
/// session it relaunched had been active this second. On the phone that
/// is every row reading "just now" and the list -- which is sorted by
/// this -- coming back in an order that means nothing, with the
/// conversation somebody was in the middle of buried among sessions
/// untouched for days.
///
/// `None` for a transcript with no lines in it, which is a session that
/// genuinely has not done anything yet; its caller uses the clock, which
/// is right there and only there.
pub fn last_activity(&self) -> Option<f64> {
self.last_activity
}
/// Everything this session has spent, as of opening.
///
/// Zero for an empty transcript, which is a session that has spent
/// nothing -- the one case where zero is the answer rather than the
/// absence of one.
pub fn total_tokens(&self) -> u64 {
self.total_tokens
}
/// Appends `event`, assigning it the next sequence number. Flushed per
/// event: each line is tiny, and the transcript is the source of truth
/// a crash must not lose the tail of.
@@ -352,7 +392,10 @@ mod tests {
Event::Status {
state: SessionStatus::Idle,
},
Event::UsageDelta { tokens: 42 },
Event::UsageDelta {
tokens: 42,
total: 0,
},
Event::Error {
message: "boom".into(),
},