perf: update chats incrementally instead of replaying their history

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 - Add `toolCall.askUser.trustModeTimeoutSeconds`: in trust mode, an unanswered `ask_user` question times out and the LLM continues with its best judgment.
+- Speed up rollback, flags and forks on long chats via the `incrementalUpdates` capability: `chat/truncated`, positioned flags, and paged `/fork`, `/btw` and `chat/fork`.
 
 ## 0.165.0
 

docs/protocol.md

@@ -235,6 +235,24 @@ interface ClientCapabilities {
              * during a chat session.
              */
             askQuestion?: boolean;
+
+            /**
+             * Present when the client applies chat history changes in place
+             * instead of needing a full `chat/cleared` + replay. Then
+             * `chat/rollback` emits `chat/truncated`, `chat/addFlag` emits a
+             * single flag `chat/contentReceived` carrying `afterContentId`, and
+             * chats the server creates (`/fork`, `/btw`, `chat/fork`) stream
+             * only their newest messages, with `chat/opened` `history`
+             * describing the rest. Remote (SSE) clients keep receiving the
+             * full replays.
+             */
+            incrementalUpdates?: {
+                /**
+                 * Max messages to stream into a chat the server creates.
+                 * Defaults to 50.
+                 */
+                historyPageSize?: number;
+            };
         }
     }
 }
@@ -1536,6 +1554,15 @@ interface ChatFlagContent {
      * Unique identifier for this flag.
      */
     contentId: string;
+
+    /**
+     * Id of the message the flag belongs after (message contentId or
+     * content id). Only sent by `chat/addFlag` to clients with the
+     * `incrementalUpdates` capability; when absent, render the flag at the
+     * current position. Clients may skip rendering when that message is not
+     * loaded (e.g. on an unloaded history page).
+     */
+    afterContentId?: string;
 }
 
 ```
@@ -1834,7 +1861,8 @@ interface ChatPromptSteerRemoveParams {
 A client request to rollback chat messages to before a specific user sent message using `contentId`.
 Clients should show an option close to user sent messages in chat to rollback, calling this method.
 Server will then remove the messages from its memory after that contentId and produce `chat/cleared` followed 
-with `chat/contentReceived` with the kept messages.
+with `chat/contentReceived` with the kept messages. Clients with the `incrementalUpdates` capability receive
+a single `chat/truncated` instead.
 
 _Request:_ 
 
@@ -1907,6 +1935,8 @@ The flag is inserted after the message identified by the given `contentId`.
 The `contentId` matches against both message-level `contentId` (user messages) and
 content `id` fields (tool calls, reasons, etc), allowing placement between any message types.
 Flags are persisted as messages and propagate automatically with fork and resume.
+Server then emits `chat/cleared` followed by the full history via `chat/contentReceived`, or, for clients with
+the `incrementalUpdates` capability, a single flag `chat/contentReceived` carrying `afterContentId`.
 
 _Request:_
 
@@ -1974,7 +2004,9 @@ interface ChatRemoveFlagResponse {}
 A client request to fork the chat into a new chat with messages up to and including
 the message identified by the given `contentId`. The server creates the new chat,
 sends a `chat/opened` notification with the new chat, then streams the kept messages
-via `chat/contentReceived`. A system message is also sent to the original chat.
+(only the newest page of them for clients with the `incrementalUpdates` capability) via
+`chat/contentReceived`. The fork always keeps the full history. A system message is
+also sent to the original chat.
 
 _Request:_
 
@@ -2004,6 +2036,12 @@ interface ChatForkResponse {
      * Absent when the given contentId was not found.
      */
     chatId?: string;
+
+    /**
+     * Pagination info for the streamed messages, present when the client has
+     * the `incrementalUpdates` capability. Same as `chat/opened` `history`.
+     */
+    meta?: ChatHistoryMeta;
 }
 ```
 
@@ -2031,12 +2069,38 @@ interface ChatClearedParams {
 }
 ```
 
+### Chat truncated (⬅️)
+
+A server notification to remove a message and everything after it from a chat UI, keeping the earlier
+messages as they are. Only sent to clients with the `incrementalUpdates` capability, currently after
+`chat/rollback`. When the message is not loaded in the client (e.g. it is on an unloaded history page),
+clients should re-open the chat via `chat/open`.
+
+_Notification:_ 
+
+* method: `chat/truncated`
+* params: `ChatTruncatedParams` defined as follows:
+
+```typescript
+interface ChatTruncatedParams {
+    /**
+     * The chat session identifier.
+     */
+    chatId: string;
+
+    /**
+     * The contentId of the first removed message.
+     */
+    contentId: string;
+}
+```
+
 ### Chat opened (⬅️)
 
 A server notification indicating that a chat exists or was just materialized
 server-side. Emitted when:
 
-- a chat is forked via `/fork` or the `chat/fork` request,
+- a chat is forked via `/fork`, `/btw` or the `chat/fork` request,
 - a persisted chat is replayed via `chat/open`,
 - a client-supplied `chatId` is seen for the first time on `chat/prompt`
   (the legacy null-`chatId` path does not emit this).
@@ -2064,6 +2128,14 @@ interface ChatOpenedParams {
      * not yet have a title.
      */
     title?: string;
+
+    /**
+     * Pagination info when only the newest page of the chat's messages
+     * follows, for chats the server creates (`/fork`, `/btw`, `chat/fork`)
+     * for clients with the `incrementalUpdates` capability. Older messages
+     * can be paged via `chat/history` using `beforeCursor`.
+     */
+    history?: ChatHistoryMeta;
 }
 ```
 
@@ -2346,8 +2418,10 @@ the selected window; an opaque `after` cursor pages forward instead. The literal
 sentinel `"lastCompaction"` may be passed as `before`/`after`: `after: "lastCompaction"`
 returns the messages since the last compaction (the active context),
 `before: "lastCompaction"` the summarized-away history. Cursors are ephemeral —
-a cursor whose message no longer exists (e.g. after a rollback or clear) returns
-a `cursor_expired` error, prompting the client to refetch the latest page.
+a cursor whose message no longer exists returns a `cursor_expired` error,
+prompting the client to refetch the latest page. The exception is a cursor whose
+message was cut off by a rollback: everything left is older than it, so a
+`before` page returns the newest remaining messages.
 
 _Request:_
 

src/eca/db.clj

@@ -41,7 +41,8 @@
    :client-capabilities {:code-assistant {:editor {:diagnostics :boolean
                                                    :definition :boolean
                                                    :references :boolean}
-                                          :chat-capabilities {:ask-question :boolean}}}
+                                          :chat-capabilities {:ask-question :boolean
+                                                              :incremental-updates {:history-page-size :number}}}}
    :config-hash :string
    :providers-config-hash :string
    :last-config-notified ::any-map

src/eca/features/chat.clj

@@ -2632,7 +2632,8 @@
 
 (defn rollback-chat
   "Remove messages from chat in db until content-id matches.
-   Then notify to clear chat and then the kept messages."
+   Then notify the client: `chat/truncated` when it supports incremental
+   updates, otherwise `chat/cleared` followed by the kept messages."
   [{:keys [chat-id content-id include]} db* messenger metrics]
   (db/hydrate-chat! db* chat-id metrics)
   (let [include (if (seq include)
@@ -2661,15 +2662,20 @@
       ;; state. Persist immediately so the cleaned-up history survives a
       ;; restart instead of relying on the next unrelated save.
       (db/save-chat! @db* chat-id metrics)
-      (messenger/chat-cleared
-       messenger
-       {:chat-id chat-id
-        :messages true})
-      (send-chat-contents!
-       new-messages
-       {:chat-id chat-id
-        :db* db*
-        :messenger messenger}))
+      (if (history/incremental-updates? @db*)
+        (when (< (count new-messages) (count all-messages))
+          (messenger/chat-truncated messenger {:chat-id chat-id
+                                               :content-id content-id}))
+        (do
+          (messenger/chat-cleared
+           messenger
+           {:chat-id chat-id
+            :messages true})
+          (send-chat-contents!
+           new-messages
+           {:chat-id chat-id
+            :db* db*
+            :messenger messenger}))))
     {}))
 
 (defn ^:private find-last-message-idx
@@ -2696,7 +2702,9 @@
   "Add a named flag after the message identified by content-id.
    Searches both :content-id and [:content :id] to support placement
    after any message type (user, tool call, reason, etc).
-   Clears and replays the chat to render the flag at the correct position."
+   Clients supporting incremental updates receive only the flag, with
+   `afterContentId` telling where to render it; others get a clear and
+   full replay so the flag renders at the correct position."
   [{:keys [chat-id content-id text]} db* messenger metrics]
   (db/hydrate-chat! db* chat-id metrics)
   (let [messages (vec (get-in @db* [:chats chat-id :messages]))
@@ -2709,8 +2717,17 @@
                                (cons flag-msg (subvec messages insert-after)))]
         (swap! db* assoc-in [:chats chat-id :messages] new-messages)
         (db/save-chat! @db* chat-id metrics)
-        (messenger/chat-cleared messenger {:chat-id chat-id :messages true})
-        (send-chat-contents! new-messages {:chat-id chat-id :db* db* :messenger messenger})))
+        (if (history/incremental-updates? @db*)
+          (messenger/chat-content-received messenger
+                                           {:chat-id chat-id
+                                            :role :system
+                                            :content {:type :flag
+                                                      :text text
+                                                      :contentId flag-id
+                                                      :afterContentId content-id}})
+          (do
+            (messenger/chat-cleared messenger {:chat-id chat-id :messages true})
+            (send-chat-contents! new-messages {:chat-id chat-id :db* db* :messenger messenger})))))
     {}))
 
 (defn remove-flag
@@ -2728,7 +2745,10 @@
 
 (defn fork-chat
   "Fork the chat creating a new chat with messages up to and including
-   the message identified by content-id."
+   the message identified by content-id.
+   Clients supporting incremental updates get only the newest page of kept
+   messages; the paging meta is returned as `:meta` and in `chat/opened` so
+   the client can page older via `chat/history`."
   [{:keys [chat-id content-id]} db* messenger metrics]
   (db/hydrate-chat! db* chat-id metrics)
   (let [chat (get-in @db* [:chats chat-id])
@@ -2748,19 +2768,23 @@
                       :model (:model chat)
                       :last-api (:last-api chat)
                       :messages kept-messages
-                      :prompt-finished? true}]
+                      :prompt-finished? true}
+            page (history/server-created-page @db* kept-messages)
+            meta (when page (history/page-meta page kept-messages))]
         (swap! db* assoc-in [:chats new-id] new-chat)
         (mark-editor-open! db* new-id)
         (db/save-chat! @db* new-id metrics)
-        (messenger/chat-opened messenger {:chat-id new-id :title new-title})
-        (send-chat-contents! kept-messages {:chat-id new-id :db* db* :messenger messenger})
+        (messenger/chat-opened messenger (assoc-some {:chat-id new-id :title new-title}
+                                                     :history meta))
+        (send-chat-contents! (if page (:messages page) kept-messages)
+                             {:chat-id new-id :db* db* :messenger messenger})
         (lifecycle/send-content! {:messenger messenger :chat-id new-id}
                                  :system
                                  (assoc-some {:type :metadata} :title new-title))
         (lifecycle/send-content! {:messenger messenger :chat-id chat-id}
                                  :system
                                  {:type :text :text (str "Chat forked to: " new-title)})
-        {:chat-id new-id}))))
+        (assoc-some {:chat-id new-id} :meta meta)))))
 
 (defn list-chats
   "Pure projection over `(:chats db)`: returns a summary list intended for the
@@ -2852,11 +2876,7 @@
                         :title title
                         :selection selection}
                        :meta (when windowed?
-                               {:total (:total page)
-                                :returned (:returned page)
-                                :before-cursor (:before-cursor page)
-                                :after-cursor (:after-cursor page)
-                                :compaction-cursor (history/compaction-cursor (:messages chat))}))))))))
+                               (history/page-meta page (:messages chat))))))))))
 
 (defn fetch-history
   "Window a chat's persisted messages and return the transformed content items
@@ -2880,8 +2900,4 @@
                     :message "Cursor no longer points to an existing message; refetch the latest page"}}
            {:contents (messages->contents (:messages result)
                                           {:chat-id chat-id :parent-chat-id nil :db db})
-            :meta {:total (:total result)
-                   :returned (:returned result)
-                   :before-cursor (:before-cursor result)
-                   :after-cursor (:after-cursor result)
-                   :compaction-cursor (history/compaction-cursor messages)}}))))))
+            :meta (history/page-meta result messages)}))))))

src/eca/features/chat/history.clj

@@ -6,7 +6,9 @@
    Cursors are opaque tokens: base64url of \"<index>.<checksum>\" where checksum
    fingerprints the message. Resolving trusts the encoded index when its checksum
    still matches, else rescans by checksum (tolerating shifts like flag
-   insert/remove); a cursor whose message is gone resolves to :expired."
+   insert/remove); a cursor whose message was cut off by truncating the history
+   resolves to its end, and any other cursor whose message is gone resolves to
+   :expired."
   (:require
    [clojure.string :as string])
   (:import
@@ -45,14 +47,19 @@
       :else (recur (dec i)))))
 
 (defn ^:private resolve-cursor
-  "Resolves an opaque cursor to an index, or :expired if its message is gone."
+  "Resolves an opaque cursor to an index, or :expired if its message is gone.
+   A gone message whose index is past the end resolves to the end, as the
+   history was truncated at or before it (rollback) and every remaining
+   message is older."
   [messages ^String cursor]
   (if-let [{:keys [idx checksum]} (decode-cursor cursor)]
-    (if (and (<= 0 idx) (< idx (count messages))
-             (= checksum (message-checksum (nth messages idx))))
-      idx
-      (or (first (keep-indexed (fn [i m] (when (= checksum (message-checksum m)) i)) messages))
-          :expired))
+    (let [n (count messages)]
+      (if (and (<= 0 idx) (< idx n)
+               (= checksum (message-checksum (nth messages idx))))
+        idx
+        (or (first (keep-indexed (fn [i m] (when (= checksum (message-checksum m)) i)) messages))
+            (when (>= idx n) n)
+            :expired)))
     :expired))
 
 (defn ^:private resolve-bound
@@ -140,3 +147,41 @@
   [messages]
   (when-let [idx (last-compaction-idx (vec messages))]
     (encode-cursor idx (nth (vec messages) idx))))
+
+(defn page-meta
+  "Pagination meta for a `window-messages` `page` of `messages`."
+  [page messages]
+  {:total (:total page)
+   :returned (:returned page)
+   :before-cursor (:before-cursor page)
+   :after-cursor (:after-cursor page)
+   :compaction-cursor (compaction-cursor messages)})
+
+(def default-history-page-size
+  "Messages sent when the server creates a chat for a client supporting
+   incremental updates that did not declare a `historyPageSize`."
+  50)
+
+(defn ^:private incremental-updates-capability [db]
+  (get-in db [:client-capabilities :code-assistant :chat-capabilities :incremental-updates]))
+
+(defn incremental-updates?
+  "Whether the client applies chat history changes in place and pages
+   server-created chats instead of needing full replays."
+  [db]
+  (boolean (incremental-updates-capability db)))
+
+(defn history-page-size
+  "Number of messages to send for a chat the server creates, as declared by
+   the client's `incrementalUpdates` capability."
+  [db]
+  (let [capability (incremental-updates-capability db)]
+    (or (coerce-limit (when (map? capability) (:history-page-size capability)))
+        default-history-page-size)))
+
+(defn server-created-page
+  "The newest page of `messages` to send for a chat the server creates, or nil
+   when the client needs the full history."
+  [db messages]
+  (when (incremental-updates? db)
+    (window-messages messages {:limit (history-page-size db)})))