diff --git a/internal/conversation/conversation.go b/internal/conversation/conversation.go index bb82dde..187c67e 100644 --- a/internal/conversation/conversation.go +++ b/internal/conversation/conversation.go @@ -155,6 +155,7 @@ type EventType string const ( EventAssistantUpdate EventType = "assistant_update" EventDone EventType = "done" + EventInferenceActive EventType = "inference_active" EventMessageDeleted EventType = "message_deleted" EventMessageSaved EventType = "message_saved" ) @@ -197,6 +198,22 @@ type Done struct { func (Done) Type() EventType { return EventDone } +// InferenceActive reports that an [Inference] is reserved or in progress. +// [Conversation.Subscribe] sends this event one time, after the cache diff +// and before the live events of the run. +// +// Without this event, two cases look the same to the client: +// - No run is active. The stream closes after the diff. +// - A run is active, but it sends no events yet. For example, the model +// has not sent its first token, or a slow tool is running. +// +// A client that did not start the run uses this event to show the stop +// control. The client also uses this event to separate saved history +// (before the event) from live events (after the event). +type InferenceActive struct{} + +func (InferenceActive) Type() EventType { return EventInferenceActive } + // Inference is one reserved inference call and its events. // [Conversation.Reserve] creates it, and [Inference.Run] runs it. // [Inference.Events] and [Conversation.Subscribe] use its events to send @@ -323,11 +340,11 @@ func (inf *Inference) follow( // 1. Cache diff: [MessageSaved] for each saved message missing from or // newer than manifest, followed by [MessageDeleted] for each ID in // manifest that is no longer saved. -// 2. Active inference: if an [Inference] is reserved or running when -// Subscribe loads the database, Subscribe streams its -// [MessageSaved] and [AssistantUpdate] events, then ends with [Done]. -// If no inference is reserved or running, Subscribe sends no [Done] -// event. +// 2. Active inference: if an [Inference] is reserved or in progress when +// this method reads the database, the stream sends [InferenceActive], +// then the [MessageSaved] and [AssistantUpdate] events of the run, and +// then [Done]. If no run is active, the stream closes after the diff and +// does not send [InferenceActive] or [Done]. // // The channel closes when delivery finishes or ctx is canceled. Canceling // ctx stops delivery to this subscriber; it does not stop an ongoing @@ -372,6 +389,9 @@ func (c *Conversation) Subscribe( if inf == nil { return } + if !send(ctx, ch, InferenceActive{}) { + return + } // Skip events for messages that the initial diff already // sent. If the client's manifest already had a message from // this call, it is not in sent and will be re-sent here; the diff --git a/internal/conversation/conversation_test.go b/internal/conversation/conversation_test.go index d166fe9..2c1497f 100644 --- a/internal/conversation/conversation_test.go +++ b/internal/conversation/conversation_test.go @@ -1474,17 +1474,21 @@ func TestSubscribeInference(t *testing.T) { close(release) got := collectEvents(t, ch) - // The diff delivers the user message first. + // The cache diff sends the user message first. Then, the active marker + // shows that the live events of the run start. saved, ok := got[0].(MessageSaved) if !ok || saved.Message.Role != db.MessageRoleUser { t.Fatalf("first event = %v, want the user MessageSaved", got[0]) } + if _, ok := got[1].(InferenceActive); !ok { + t.Fatalf("second event = %v, want InferenceActive", got[1]) + } var ( chunkTexts []string chunkIDs []string asst *MessageSaved ) - for _, e := range got[1:] { + for _, e := range got[2:] { switch e := e.(type) { case AssistantUpdate: if asst != nil { @@ -1589,6 +1593,10 @@ func TestSubscribeInferenceError(t *testing.T) { close(release) got := collectEvents(t, ch) + // The active marker comes after the user message from the cache diff. + if _, ok := got[1].(InferenceActive); !ok { + t.Fatalf("second event = %v, want InferenceActive", got[1]) + } done, ok := got[len(got)-1].(Done) if !ok { t.Fatalf("last event = %v, want Done", got[len(got)-1]) @@ -1709,12 +1717,18 @@ func TestSubscribeToolResult(t *testing.T) { t.Errorf("event %d role = %q, want the user first", i, e.Message.Role) } } + case InferenceActive: + // This event separates the cache diff from the live events + // of the run. case Done: if e.Error != "" { t.Errorf("Done.Error = %q, want empty", e.Error) } default: - t.Errorf("event %d = %v, want MessageSaved or Done", i, e) + t.Errorf( + "event %d = %v, want MessageSaved, InferenceActive, or Done", + i, e, + ) } } if aId == "" { diff --git a/internal/service/static/js/api.js b/internal/service/static/js/api.js index 68aa8d5..1460359 100644 --- a/internal/service/static/js/api.js +++ b/internal/service/static/js/api.js @@ -165,9 +165,10 @@ export class Api { * message_deleted per manifest ID that is no longer saved. An * empty manifest requests every saved row. * - * When an inference is reserved or ongoing, the stream then follows - * its events and ends with done. When none is, the stream ends - * after the diff without done. + * If an inference run is reserved or in progress, the stream sends one + * inference_active event, then the live events of the run, and then done. + * If no run is active, the stream closes after the diff and does not send + * inference_active or done. * * @param {string} id - conversation ID. * @param {Array<{id: string, updated_at: string}>} manifest @@ -345,7 +346,10 @@ export class Api { body, signal, }); - yield* readEvents(res); + for await (const event of readEvents(res)) { + if (signal?.aborted) return; + yield event; + } } } diff --git a/internal/service/static/js/main.js b/internal/service/static/js/main.js index 75a090d..51e897b 100644 --- a/internal/service/static/js/main.js +++ b/internal/service/static/js/main.js @@ -200,28 +200,15 @@ class Odidere { // ==================== /** - * orderedRows returns the cached rows in the order they were - * saved. The server sends rows in saved order on every stream, - * and the map keeps that order through updates. + * orderedRows returns the cached rows in saved order, sorted by + * sequence. The cache is a Map keyed by message ID, and a Map + * keeps the order in which keys were first added: a row that + * another client inserted earlier in the conversation arrives + * after rows that follow it. Sorting puts it in its place. * @returns {Array} */ orderedRows() { - return [...this.messages.values()]; - } - - /** - * isOrdered reports whether the cached rows are still in saved - * order. A row another client inserted or deleted can shift the - * sequences of rows the sync diff does not re-send, because their - * updated_at did not change; the check catches that. - */ - isOrdered() { - let previous = -Infinity; - for (const row of this.messages.values()) { - if (row.sequence <= previous) return false; - previous = row.sequence; - } - return true; + return [...this.messages.values()].sort((a, b) => a.sequence - b.sequence); } /** @@ -329,31 +316,36 @@ class Odidere { * @param {string} type * @param {Object} data * @param {string} [voice] + * @param {boolean} [speak] - false while reading the saved diff, + * so history is not read aloud. * @returns {boolean} true when the stream is finished. */ - handleEvent(type, data, voice = '') { + handleEvent(type, data, voice = '', speak = true) { switch (type) { case 'message_saved': { const row = data.message; - const existed = this.messages.has(row.id); + const previous = this.messages.get(row.id); + // When a subscriber joins a run, the server resends the + // run's events for messages the client's manifest already + // held (see Subscribe in conversation.go). Skip a revision + // that is already cached, so it does not render or speak + // again. + if (previous?.updated_at === row.updated_at) return false; + const existed = previous !== undefined; this.cacheRow(row, voice); - this.applySavedRow(row, existed, voice); + this.applySavedRow(row, existed, voice, speak); this.trackUsage(row.metadata?.usage); return false; } case 'message_deleted': { - const row = this.messages.get(data.message_id); - this.messages.delete(data.message_id); + const existed = this.messages.delete(data.message_id); this.persistCache(); - if (!row) return false; - if (row.role === 'tool') { - this.ui.clearToolOutput(row); - } else { - this.ui.removeMessageBubbles(data.message_id); - } + if (existed) this.ui.renderConversation(this.orderedRows()); return false; } case 'assistant_update': { + // Skip resent chunks of a message that is already saved. + if (this.messages.has(data.message_id)) return false; const draft = this.applyChunk(data.message_id, data.chunk); this.ui.renderDraft(draft); this.trackUsage(data.chunk?.usage); @@ -376,8 +368,9 @@ class Odidere { * @param {Object} row * @param {boolean} existed - whether the row was already cached. * @param {string} voice - the voice of this run, '' when unknown. + * @param {boolean} [speak] - whether to read the reply aloud. */ - applySavedRow(row, existed, voice = '') { + applySavedRow(row, existed, voice = '', speak = true) { const wasDraft = this.drafts.delete(row.id); if (row.role === 'tool') { this.ui.attachToolResult( @@ -387,12 +380,11 @@ class Odidere { ); return; } - if (existed || wasDraft) { - if (wasDraft) { - this.ui.replaceDraftWithSaved(row.id, row, this.resultsFor(row)); - } else { - this.ui.renderConversation(this.orderedRows()); - } + const rows = this.orderedRows(); + if (wasDraft) { + this.ui.replaceDraftWithSaved(row.id, row, this.resultsFor(row)); + } else if (existed || rows.at(-1)?.id !== row.id) { + this.ui.renderConversation(rows); } else { this.ui.renderMessages([row, ...this.resultsFor(row)]); if (row.role === 'user') { @@ -405,7 +397,7 @@ class Odidere { } } } - if (row.role === 'assistant') { + if (speak && row.role === 'assistant') { const text = extractText(row.content?.content); const cached = this.messages.get(row.id); if (text) { @@ -469,6 +461,25 @@ class Odidere { this.drafts.clear(); } + /** + * setBusy marks the conversation busy while a run or an attached + * subscription streams, or clears that state and discards unsaved + * drafts when the stream ends. + * @param {boolean} busy + */ + setBusy(busy) { + this.busy = busy; + this.ui.setBusy(busy); + if (busy) { + this.ui.setPendingApprovals([]); + this.wakeLock.acquire(); + } else { + this.clearDrafts(); + this.updateSendEnabled(); + this.wakeLock.release(); + } + } + /** * trackUsage keeps the newest reported token total. Streaming * usage reports replace earlier ones; they are not increments. @@ -568,11 +579,7 @@ class Odidere { const controller = new AbortController(); this.runController = controller; - this.busy = true; - this.ui.setBusy(true); - this.ui.setPendingApprovals([]); - // Keep the screen on while the run streams. - this.wakeLock.acquire(); + this.setBusy(true); // Consume the input once the server accepted the request: an // earlier failure keeps it for a retry. @@ -630,12 +637,10 @@ class Odidere { // An AbortError ends delivery only; reset deleted the // conversation and will clear the local state. } finally { - if (this.runController === controller) this.runController = null; - this.clearDrafts(); - this.busy = false; - this.ui.setBusy(false); - this.updateSendEnabled(); - this.wakeLock.release(); + if (this.runController === controller) { + this.runController = null; + this.setBusy(false); + } if (this.conversationId && !controller.signal.aborted) { await this.sync(); } @@ -776,6 +781,7 @@ class Odidere { await this.currentRun; } catch {} } + if (this.busy) this.setBusy(false); } // ==================== @@ -856,9 +862,10 @@ class Odidere { await this.currentRun; } catch {} } + if (this.busy) this.setBusy(false); } - // Drop the local state of the old selection. resync reads and + // Drop the local state of the old selection. sync reads and // renders the whole history of the new selection and caches it. this.messages.clear(); this.clearDrafts(); @@ -871,7 +878,7 @@ class Odidere { this.ui.updateContext(0, null); this.conversationId = id; this.ui.selectComposeModel(this.resolveConversationModel(id)); - await this.resync(); + await this.sync(); this.updateSendEnabled(); } @@ -943,19 +950,21 @@ class Odidere { /** * sync subscribes with the cached rows as its manifest. The - * server sends what changed since they were saved, then the - * events of a run in progress, ending with done. Without a run - * in progress, the stream ends after the diff: the cache is then - * in step with the saved history until the next change. + * server sends what changed since they were saved, then, when a + * run is in progress, inference_active and the run's events, + * ending with done. Without a run in progress, the stream ends + * after the diff: the cache is then in step with the saved + * history until the next change. * * Re-subscribing is the reconnect path: it replays what a lost * stream missed and re-attaches to a run still in progress. */ async sync() { - if (!this.conversationId || this.busy) return; + if (!this.conversationId || this.runController) return; const controller = new AbortController(); this.syncController?.abort(); this.syncController = controller; + let live = false; try { const manifest = this.orderedRows().map(({ id, updated_at }) => ({ id, @@ -966,9 +975,13 @@ class Odidere { manifest, controller.signal, )) { - if (this.handleEvent(type, data)) break; + if (type === 'inference_active') { + live = true; + this.setBusy(true); + } else if (this.handleEvent(type, data, '', live)) { + break; + } } - if (!this.isOrdered()) await this.resync(); } catch (error) { if (error instanceof ApiError && error.status === 404) { this.ui.renderError('This conversation was deleted.'); @@ -980,48 +993,12 @@ class Odidere { console.error('sync failed:', error); } } finally { - if (this.syncController === controller) this.syncController = null; - } - if (this.conversationId && !this.busy) { - await this.updateApprovals(); - } - } - - /** - * resync replaces the whole cache from the saved history. A - * structural change can shift rows the sync diff does not - * re-send, because their updated_at did not change; a full read - * restores the saved order. - */ - async resync() { - if (!this.conversationId || this.busy) return; - const controller = new AbortController(); - this.syncController?.abort(); - this.syncController = controller; - this.messages.clear(); - this.drafts.clear(); - try { - for await (const { type, data } of this.api.subscribe( - this.conversationId, - [], - controller.signal, - )) { - if (type === 'message_saved') { - this.cacheRow(data.message, '', false); - } else if (this.handleEvent(type, data)) { - break; - } + if (this.syncController === controller) { + this.syncController = null; + if (live) this.setBusy(false); } - } catch (error) { - if (error.name !== 'AbortError') { - console.error('resync failed:', error); - } - } finally { - if (this.syncController === controller) this.syncController = null; - this.persistCache(); } - this.ui.renderConversation(this.orderedRows()); - if (this.conversationId && !this.busy) { + if (!controller.signal.aborted && this.conversationId && !this.busy) { await this.updateApprovals(); } } @@ -1102,8 +1079,7 @@ class Odidere { * so the conversation can continue. The decision alone would * leave the call without a result, and every later run would * refuse the batch with 'will not retry'. The note inserts a - * message, so the cache is replaced from the saved order before - * the run continues. + * message, so sync updates the cache before the run continues. * @param {{executionId: string, messageId: string, toolCallId: string}} execution */ async deny({ executionId, messageId, toolCallId }) { @@ -1137,7 +1113,7 @@ class Odidere { this.reportMutationError(error); return; } - await this.resync(); + await this.sync(); await this.send({ text: '', voice: this.lastVoice }); } @@ -1261,8 +1237,8 @@ class Odidere { } if (!resultRow && edited.result) { - // A new result is an inserted message; replace the cache - // from the saved order before the display renders it. + // A new result is an inserted message; sync brings in the + // result and any shifted sequences. try { await this.api.insertMessage(this.conversationId, { message: { @@ -1277,7 +1253,7 @@ class Odidere { this.reportMutationError(error); return; } - await this.resync(); + await this.sync(); return; } @@ -1323,7 +1299,7 @@ class Odidere { if (field === 'tool' && toolCallId) { // The server removes the call, its result, and its // execution record together, and rewrites the assistant row; - // replace the cache from the saved order. + // sync applies those changes. try { await this.api.deleteToolCall( this.conversationId, @@ -1335,7 +1311,7 @@ class Odidere { this.reportMutationError(error); return; } - await this.resync(); + await this.sync(); return; } @@ -1354,7 +1330,7 @@ class Odidere { this.reportMutationError(error); return; } - await this.resync(); + await this.sync(); } return; } @@ -1365,13 +1341,13 @@ class Odidere { this.reportMutationError(error); return; } - await this.resync(); + await this.sync(); } /** * putMessage updates one saved message and re-renders the chat - * from the cache. An update does not shift rows, so no resync is - * needed. + * from the cache. The response returns the updated row, so no + * sync is needed. * @param {string} mid * @param {Object} body - message and updated_at. */ @@ -1399,10 +1375,10 @@ class Odidere { async reportMutationError(error) { if (error instanceof ApiError && error.status === 409) { this.ui.renderError('The message changed elsewhere. Reloading it.'); - await this.resync(); + await this.sync(); } else if (error instanceof ApiError && error.status === 404) { this.ui.renderError('The message is gone. Reloading the conversation.'); - await this.resync(); + await this.sync(); } else { this.ui.renderError(`Edit failed: ${error.message}`); }