Fetch the transcript instead of replaying it one event at a time
**The five seconds of loading top-down.** Opening a session subscribed to the event stream from sequence zero, so the backlog arrived as one SSE frame per event -- 864 of them for an imported conversation, rendered as they landed. That is not a slow list; it is a conversation being replayed at network speed, and it looks like loading from the top because it is. The newest page now comes as one request, and the stream starts from where that page ended, carrying live events only -- which is what a stream is good at. Scrolling back fetches the page before it, so history costs something only when somebody actually reads it. 80 events instead of 864, and the first frame is already the end of the conversation. I had called this fixed after anchoring the list at the bottom, on the strength of an emulator on the same machine as the server. That test could not have shown the problem: the whole backlog arrived in one frame's worth of time over loopback. Iris's phone, over a tunnel, took five seconds. **Send disappearing while running.** It was never conditional -- the row simply ran out of width. A Row hands out intrinsic widths in order and clips the overflow, so when Stop appeared the pickers I had added pushed Send off the screen: the app's central control, gone at exactly the moment the app is most in use. The settings now share what is left after the actions have taken what they need. While a turn is in flight the button says **Queue**, because that is what sending then does -- the message is injected at the next tool boundary rather than starting a turn of its own. The backend has always done this; the button was describing something else. **And the model picker no longer dismisses the keyboard**, which it did by taking focus. Changing the model mid-sentence is an aside, not a departure from what you were typing. Verified on the 864-event import: at the newest message within a second, history paging back continuously past the first page, and Stop beside Queue while running.
This commit is contained in:
1 parent
ba25a5cacf
commit
dcb158ee44
4 files changed
+190
-9
No files matched your search
@@ -439,6 +439,31 @@ fun deleteImportable(settings: ServerSettings, setup: String, sessionId: String)
|
|||||||
requestFromServer(settings, "/setups/$setup/importable/$sessionId", method = "DELETE") {}
|
requestFromServer(settings, "/setups/$setup/importable/$sessionId", method = "DELETE") {}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A page of a session's transcript, oldest first within the page.
|
||||||
|
*
|
||||||
|
* One request instead of one stream frame per event. The SSE stream is the right shape for live
|
||||||
|
* events and the wrong one for a backlog: opening an imported session replayed hundreds of frames
|
||||||
|
* before anything was readable, which looked exactly like the app loading top-down, because it was.
|
||||||
|
*
|
||||||
|
* [before] pages backwards for history somebody scrolls to; absent means the newest page.
|
||||||
|
*/
|
||||||
|
fun fetchTranscript(
|
||||||
|
settings: ServerSettings,
|
||||||
|
sessionId: String,
|
||||||
|
before: Long? = null,
|
||||||
|
limit: Int = 80,
|
||||||
|
): List<SeqEvent> {
|
||||||
|
val query = buildString {
|
||||||
|
append("?limit=").append(limit)
|
||||||
|
if (before != null) append("&before=").append(before)
|
||||||
|
}
|
||||||
|
return requestFromServer(settings, "/sessions/$sessionId/transcript$query") { connection ->
|
||||||
|
val body = JSONArray(connection.inputStream.bufferedReader().readText())
|
||||||
|
(0 until body.length()).map { parseSeqEvent(body.getJSONObject(it).toString()) }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/** Switches a running session's model; the CLI changes it in place. */
|
/** Switches a running session's model; the CLI changes it in place. */
|
||||||
fun setSessionModel(settings: ServerSettings, sessionId: String, model: String) {
|
fun setSessionModel(settings: ServerSettings, sessionId: String, model: String) {
|
||||||
requestFromServer(
|
requestFromServer(
|
||||||
|
|||||||
@@ -47,7 +47,9 @@ import androidx.compose.ui.Modifier
|
|||||||
import androidx.compose.ui.graphics.ImageBitmap
|
import androidx.compose.ui.graphics.ImageBitmap
|
||||||
import androidx.compose.ui.graphics.asImageBitmap
|
import androidx.compose.ui.graphics.asImageBitmap
|
||||||
import androidx.compose.ui.platform.LocalContext
|
import androidx.compose.ui.platform.LocalContext
|
||||||
|
import androidx.compose.ui.text.style.TextOverflow
|
||||||
import androidx.compose.ui.unit.dp
|
import androidx.compose.ui.unit.dp
|
||||||
|
import androidx.compose.ui.window.PopupProperties
|
||||||
import java.util.concurrent.atomic.AtomicLong
|
import java.util.concurrent.atomic.AtomicLong
|
||||||
import java.util.concurrent.atomic.AtomicReference
|
import java.util.concurrent.atomic.AtomicReference
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
@@ -161,6 +163,14 @@ fun SessionScreen(
|
|||||||
// The resume cursor, written from the stream's IO thread.
|
// The resume cursor, written from the stream's IO thread.
|
||||||
val lastSeq = remember { AtomicLong(0) }
|
val lastSeq = remember { AtomicLong(0) }
|
||||||
val activeStream = remember { AtomicReference<EventStream?>(null) }
|
val activeStream = remember { AtomicReference<EventStream?>(null) }
|
||||||
|
// The oldest sequence number loaded, and whether there is more behind
|
||||||
|
// it. Paging backwards is what keeps opening a long session cheap: the
|
||||||
|
// screen starts with the end of the conversation and fetches earlier
|
||||||
|
// pages only when somebody scrolls to them.
|
||||||
|
var oldestSeq by remember { mutableLongStateOf(0L) }
|
||||||
|
var moreHistory by remember { mutableStateOf(true) }
|
||||||
|
var loadingHistory by remember { mutableStateOf(false) }
|
||||||
|
var ready by remember { mutableStateOf(false) }
|
||||||
val listState = rememberLazyListState()
|
val listState = rememberLazyListState()
|
||||||
|
|
||||||
fun apply(entry: SeqEvent) {
|
fun apply(entry: SeqEvent) {
|
||||||
@@ -175,7 +185,25 @@ fun SessionScreen(
|
|||||||
// The stream lifecycle: connect, follow, and on any drop reconnect
|
// The stream lifecycle: connect, follow, and on any drop reconnect
|
||||||
// from the cursor -- so a flaky link (or a backend restart) costs
|
// from the cursor -- so a flaky link (or a backend restart) costs
|
||||||
// nothing but the gap's latency.
|
// nothing but the gap's latency.
|
||||||
|
// The newest page first, in one request, before the stream opens. The
|
||||||
|
// stream then starts from where that page ended, so it carries live
|
||||||
|
// events only -- which is what it is good at.
|
||||||
LaunchedEffect(summary.id) {
|
LaunchedEffect(summary.id) {
|
||||||
|
try {
|
||||||
|
val page = withContext(Dispatchers.IO) { fetchTranscript(settings, summary.id) }
|
||||||
|
page.forEach { apply(it) }
|
||||||
|
oldestSeq = page.firstOrNull()?.seq ?: 0L
|
||||||
|
moreHistory = oldestSeq > 1L
|
||||||
|
} catch (e: ApiException) {
|
||||||
|
// Not fatal: the stream below still replays from zero, which is
|
||||||
|
// slow but complete. Saying so beats silently showing nothing.
|
||||||
|
streamError = e.message
|
||||||
|
}
|
||||||
|
ready = true
|
||||||
|
}
|
||||||
|
|
||||||
|
LaunchedEffect(summary.id, ready) {
|
||||||
|
if (!ready) return@LaunchedEffect
|
||||||
while (true) {
|
while (true) {
|
||||||
val stream = EventStream(settings, summary.id)
|
val stream = EventStream(settings, summary.id)
|
||||||
activeStream.set(stream)
|
activeStream.set(stream)
|
||||||
@@ -217,6 +245,47 @@ fun SessionScreen(
|
|||||||
snapshotFlow { items.size }
|
snapshotFlow { items.size }
|
||||||
.collect { count -> if (followTail && count > 0) listState.scrollToItem(0) }
|
.collect { count -> if (followTail && count > 0) listState.scrollToItem(0) }
|
||||||
}
|
}
|
||||||
|
// Reaching the far end of what is loaded -- the oldest item, which in
|
||||||
|
// this layout is the last index -- fetches the page before it.
|
||||||
|
LaunchedEffect(listState, items.size, moreHistory) {
|
||||||
|
snapshotFlow { listState.layoutInfo.visibleItemsInfo.lastOrNull()?.index ?: 0 }
|
||||||
|
.collect { last ->
|
||||||
|
if (!moreHistory || loadingHistory || items.isEmpty()) return@collect
|
||||||
|
if (last < items.size - 3) return@collect
|
||||||
|
loadingHistory = true
|
||||||
|
try {
|
||||||
|
val older =
|
||||||
|
withContext(Dispatchers.IO) {
|
||||||
|
fetchTranscript(settings, summary.id, before = oldestSeq)
|
||||||
|
}
|
||||||
|
if (older.isEmpty()) {
|
||||||
|
moreHistory = false
|
||||||
|
} else {
|
||||||
|
oldestSeq = older.first().seq
|
||||||
|
moreHistory = oldestSeq > 1L
|
||||||
|
// Folded oldest-first into a list of their own, then
|
||||||
|
// put in front: `foldEvent` merges streaming text
|
||||||
|
// into the item before it, so replaying an older page
|
||||||
|
// through the live list would glue it onto the newest
|
||||||
|
// message rather than its own.
|
||||||
|
var earlier = listOf<TranscriptItem>()
|
||||||
|
older.forEach { entry ->
|
||||||
|
val event = entry.event
|
||||||
|
if (
|
||||||
|
event !is SessionEvent.Status && event !is SessionEvent.UsageDelta
|
||||||
|
) {
|
||||||
|
earlier = foldEvent(earlier, event)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
items = earlier + items
|
||||||
|
}
|
||||||
|
} catch (_: ApiException) {
|
||||||
|
// Leave `moreHistory` alone: the next scroll asks again.
|
||||||
|
} finally {
|
||||||
|
loadingHistory = false
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
LaunchedEffect(summary.setupName, summary.provider) {
|
LaunchedEffect(summary.setupName, summary.provider) {
|
||||||
offeredModels =
|
offeredModels =
|
||||||
@@ -236,6 +305,8 @@ fun SessionScreen(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
val running = status == "running" || status == "compacting"
|
||||||
|
|
||||||
fun act(action: () -> Unit) {
|
fun act(action: () -> Unit) {
|
||||||
scope.launch {
|
scope.launch {
|
||||||
try {
|
try {
|
||||||
@@ -411,10 +482,16 @@ fun SessionScreen(
|
|||||||
) {
|
) {
|
||||||
Text(if (pendingAttachments.isEmpty()) "+" else "+${pendingAttachments.size}")
|
Text(if (pendingAttachments.isEmpty()) "+" else "+${pendingAttachments.size}")
|
||||||
}
|
}
|
||||||
// Beside the field they govern, and showing their current
|
// The settings share what is left after the actions have
|
||||||
// value rather than a label: what this session is set to is
|
// taken what they need. A Row hands out intrinsic widths in
|
||||||
// the thing worth reading at a glance, and the control for
|
// order and clips whatever runs past the edge, so with
|
||||||
// changing it is the same object.
|
// these laid out first the arrival of Stop pushed Send off
|
||||||
|
// the screen entirely -- the app's central control, gone at
|
||||||
|
// exactly the moment the app is most in use.
|
||||||
|
Row(
|
||||||
|
verticalAlignment = Alignment.CenterVertically,
|
||||||
|
modifier = Modifier.weight(1f),
|
||||||
|
) {
|
||||||
if (offeredModels.isNotEmpty()) {
|
if (offeredModels.isNotEmpty()) {
|
||||||
PickerButton(
|
PickerButton(
|
||||||
current = model ?: "default",
|
current = model ?: "default",
|
||||||
@@ -433,14 +510,19 @@ fun SessionScreen(
|
|||||||
act { setSessionPermissionMode(settings, summary.id, chosen) }
|
act { setSessionPermissionMode(settings, summary.id, chosen) }
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
Spacer(Modifier.weight(1f))
|
}
|
||||||
if (status == "running" || status == "compacting") {
|
if (running) {
|
||||||
OutlinedButton(onClick = { act { interruptSession(settings, summary.id) } }) {
|
OutlinedButton(onClick = { act { interruptSession(settings, summary.id) } }) {
|
||||||
Text("Stop")
|
Text("Stop")
|
||||||
}
|
}
|
||||||
Spacer(Modifier.width(8.dp))
|
Spacer(Modifier.width(8.dp))
|
||||||
}
|
}
|
||||||
Button(onClick = { send() }) { Text("Send") }
|
// "Queue" while a turn is in flight, because that is what
|
||||||
|
// sending then does: the message is injected at the next
|
||||||
|
// tool boundary rather than starting a turn of its own.
|
||||||
|
// Naming it Send there would promise something immediate
|
||||||
|
// and describe something that waits.
|
||||||
|
Button(onClick = { send() }) { Text(if (running) "Queue" else "Send") }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -571,9 +653,24 @@ private fun PickerButton(current: String, options: List<String>, onPick: (String
|
|||||||
var open by remember { mutableStateOf(false) }
|
var open by remember { mutableStateOf(false) }
|
||||||
Box {
|
Box {
|
||||||
TextButton(onClick = { open = true }) {
|
TextButton(onClick = { open = true }) {
|
||||||
Text(current, style = MaterialTheme.typography.bodySmall)
|
// One line, truncated rather than wrapped: this sits in a row
|
||||||
|
// whose height is the buttons beside it, and a second line
|
||||||
|
// would move them.
|
||||||
|
Text(
|
||||||
|
current,
|
||||||
|
style = MaterialTheme.typography.bodySmall,
|
||||||
|
maxLines = 1,
|
||||||
|
overflow = TextOverflow.Ellipsis,
|
||||||
|
)
|
||||||
}
|
}
|
||||||
DropdownMenu(expanded = open, onDismissRequest = { open = false }) {
|
// Not focusable, so opening it does not take focus from the message
|
||||||
|
// field and dismiss the keyboard. Changing the model mid-sentence
|
||||||
|
// is an aside, not a departure from what you were typing.
|
||||||
|
DropdownMenu(
|
||||||
|
expanded = open,
|
||||||
|
onDismissRequest = { open = false },
|
||||||
|
properties = PopupProperties(focusable = false),
|
||||||
|
) {
|
||||||
options.forEach { option ->
|
options.forEach { option ->
|
||||||
DropdownMenuItem(
|
DropdownMenuItem(
|
||||||
text = { Text(option) },
|
text = { Text(option) },
|
||||||
|
|||||||
@@ -64,6 +64,7 @@ pub fn router(manager: Arc<SessionManager>) -> Router {
|
|||||||
.route("/sessions", get(list_sessions).post(spawn_session))
|
.route("/sessions", get(list_sessions).post(spawn_session))
|
||||||
.route("/sessions/{id}", delete(delete_session))
|
.route("/sessions/{id}", delete(delete_session))
|
||||||
.route("/sessions/{id}/events", get(events))
|
.route("/sessions/{id}/events", get(events))
|
||||||
|
.route("/sessions/{id}/transcript", get(transcript))
|
||||||
.route("/sessions/{id}/message", post(message))
|
.route("/sessions/{id}/message", post(message))
|
||||||
.route("/sessions/{id}/answer", post(answer))
|
.route("/sessions/{id}/answer", post(answer))
|
||||||
.route("/sessions/{id}/interrupt", post(interrupt))
|
.route("/sessions/{id}/interrupt", post(interrupt))
|
||||||
@@ -720,6 +721,41 @@ struct EventsQuery {
|
|||||||
/// cursor from the transcript, then live events as they happen. An SSE
|
/// cursor from the transcript, then live events as they happen. An SSE
|
||||||
/// auto-reconnect sends the last event id it saw as `Last-Event-ID`, which
|
/// auto-reconnect sends the last event id it saw as `Last-Event-ID`, which
|
||||||
/// takes precedence over `after` -- same cursor, native mechanism.
|
/// takes precedence over `after` -- same cursor, native mechanism.
|
||||||
|
#[derive(Deserialize)]
|
||||||
|
#[serde(rename_all = "camelCase")]
|
||||||
|
struct TranscriptQuery {
|
||||||
|
/// Page backwards from this sequence number; absent means the newest.
|
||||||
|
#[serde(default)]
|
||||||
|
before: Option<u64>,
|
||||||
|
#[serde(default = "default_window")]
|
||||||
|
limit: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
fn default_window() -> usize {
|
||||||
|
80
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A page of a session's transcript, newest first to open with.
|
||||||
|
///
|
||||||
|
/// One request rather than one stream frame per event. The SSE stream
|
||||||
|
/// stays as it is and remains the right shape for *live* events, which
|
||||||
|
/// arrive one at a time by nature; it is only the backlog that has to
|
||||||
|
/// stop pretending to be live.
|
||||||
|
async fn transcript(
|
||||||
|
State(manager): State<Arc<SessionManager>>,
|
||||||
|
UrlPath(id): UrlPath<String>,
|
||||||
|
Query(query): Query<TranscriptQuery>,
|
||||||
|
) -> Result<axum::Json<Vec<crate::session::transcript::SeqEvent>>, ApiError> {
|
||||||
|
let session = lookup(&manager, &id)?;
|
||||||
|
let events = crate::session::transcript::read_window(
|
||||||
|
session.transcript_path(),
|
||||||
|
query.before,
|
||||||
|
query.limit,
|
||||||
|
)
|
||||||
|
.map_err(bad_request)?;
|
||||||
|
Ok(axum::Json(events))
|
||||||
|
}
|
||||||
|
|
||||||
async fn events(
|
async fn events(
|
||||||
State(manager): State<Arc<SessionManager>>,
|
State(manager): State<Arc<SessionManager>>,
|
||||||
UrlPath(id): UrlPath<String>,
|
UrlPath(id): UrlPath<String>,
|
||||||
|
|||||||
@@ -73,6 +73,29 @@ impl Transcript {
|
|||||||
/// Replays every event with `seq > after`, oldest first. A missing file is
|
/// Replays every event with `seq > after`, oldest first. A missing file is
|
||||||
/// an empty transcript, not an error -- the session just hasn't produced an
|
/// an empty transcript, not an error -- the session just hasn't produced an
|
||||||
/// event yet.
|
/// event yet.
|
||||||
|
/// A window of the transcript ending just before `before`, newest-biased.
|
||||||
|
///
|
||||||
|
/// The screen opens on the end of a conversation, not the start of it, and
|
||||||
|
/// the end is all it can show at once. Replaying the whole file to get
|
||||||
|
/// there costs one network frame per event -- on an 863-event import that
|
||||||
|
/// was several seconds of messages arriving oldest-first, which reads as
|
||||||
|
/// the app loading top-down because that is exactly what it was doing.
|
||||||
|
///
|
||||||
|
/// `before` pages backwards for history somebody actually scrolls to; the
|
||||||
|
/// file is read whole each time because a transcript is small and a
|
||||||
|
/// seek-backwards reader would be a lot of machinery for a list that fits
|
||||||
|
/// in memory anyway.
|
||||||
|
pub fn read_window(path: &Path, before: Option<u64>, limit: usize) -> Result<Vec<SeqEvent>> {
|
||||||
|
let mut all = read_after(path, 0)?;
|
||||||
|
if let Some(before) = before {
|
||||||
|
all.retain(|entry| entry.seq < before);
|
||||||
|
}
|
||||||
|
if all.len() > limit {
|
||||||
|
all.drain(..all.len() - limit);
|
||||||
|
}
|
||||||
|
Ok(all)
|
||||||
|
}
|
||||||
|
|
||||||
pub fn read_after(path: &Path, after: u64) -> Result<Vec<SeqEvent>> {
|
pub fn read_after(path: &Path, after: u64) -> Result<Vec<SeqEvent>> {
|
||||||
let file = match File::open(path) {
|
let file = match File::open(path) {
|
||||||
Ok(file) => file,
|
Ok(file) => file,
|
||||||
|
|||||||
Reference in new issue
Block a user