Fix joining active inference

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
+//  2. Active inference: if an [Inference] is reserved or in progress when
-//     Subscribe loads the database, Subscribe streams its
+//     this method reads the database, the stream sends [InferenceActive],
-//     [MessageSaved] and [AssistantUpdate] events, then ends with [Done].
+//     then the [MessageSaved] and [AssistantUpdate] events of the run, and
-//     If no inference is reserved or running, Subscribe sends no [Done]
+//     then [Done]. If no run is active, the stream closes after the diff and
-//     event.
+//     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

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 == "" {

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
+   * If an inference run is reserved or in progress, the stream sends one
-   * its events and ends with done. When none is, the stream ends
+   * inference_active event, then the live events of the run, and then done.
-   * after the diff without 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;
+    }
   }
 }
 

internal/service/static/js/main.js

@@ -200,28 +200,15 @@ class Odidere {
   // ====================
 
   /**
-   * orderedRows returns the cached rows in the order they were
+   * orderedRows returns the cached rows in saved order, sorted by
-   * saved. The server sends rows in saved order on every stream,
+   * sequence. The cache is a Map keyed by message ID, and a Map
-   * and the map keeps that order through updates.
+   * 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<Object>}
    */
   orderedRows() {
-    return [...this.messages.values()];
+    return [...this.messages.values()].sort((a, b) => a.sequence - b.sequence);
-  }
-
-  /**
-   * 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;
   }
 
   /**
@@ -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);
+        const existed = this.messages.delete(data.message_id);
-        this.messages.delete(data.message_id);
         this.persistCache();
-        if (!row) return false;
+        if (existed) this.ui.renderConversation(this.orderedRows());
-        if (row.role === 'tool') {
-          this.ui.clearToolOutput(row);
-        } else {
-          this.ui.removeMessageBubbles(data.message_id);
-        }
         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) {
+    const rows = this.orderedRows();
     if (wasDraft) {
       this.ui.replaceDraftWithSaved(row.id, row, this.resultsFor(row));
-      } else {
+    } else if (existed || rows.at(-1)?.id !== row.id) {
-        this.ui.renderConversation(this.orderedRows());
+      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.setBusy(true);
-    this.ui.setBusy(true);
-    this.ui.setPendingApprovals([]);
-    // Keep the screen on while the run streams.
-    this.wakeLock.acquire();
 
     // 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;
+      if (this.runController === controller) {
-      this.clearDrafts();
+        this.runController = null;
-      this.busy = false;
+        this.setBusy(false);
-      this.ui.setBusy(false);
+      }
-      this.updateSendEnabled();
-      this.wakeLock.release();
       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
+   * server sends what changed since they were saved, then, when a
-   * events of a run in progress, ending with done. Without a run
+   * run is in progress, inference_active and the run's events,
-   * in progress, the stream ends after the diff: the cache is then
+   * ending with done. Without a run in progress, the stream ends
-   * in step with the saved history until the next change.
+   * 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.syncController === controller) {
-    }
+        this.syncController = null;
-    if (this.conversationId && !this.busy) {
+        if (live) this.setBusy(false);
-      await this.updateApprovals();
       }
     }
-
+    if (!controller.signal.aborted && this.conversationId && !this.busy) {
-  /**
-   * 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;
-        }
-      }
-    } 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) {
       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
+   * message, so sync updates the cache before the run continues.
-   * 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
+      // A new result is an inserted message; sync brings in the
-      // from the saved order before the display renders it.
+      // 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
+   * from the cache. The response returns the updated row, so no
-   * needed.
+   * 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}`);
     }