From 838b689c7c05bbd4838d649f0983985afccbf6f3 Mon Sep 17 00:00:00 2001
From: jensenojs <jensenojs@qq.com>
Date: Sat, 24 Oct 2026 22:42:19 +0800
Subject: [PATCH 1/3] fix(v2): let a second load_older wait for the page in
 flight

`[u`/`]u` called `load_older` on every press and the second call hit the
re-entry guard, which surfaced `load_older is already in progress`. A caller
asking for the same cursor now shares the page being fetched; a different
cursor queues behind it and rereads `_v2_older_cursor` when its turn comes.

A page landing after the snapshot it was cut from has been replaced is dropped
instead of restarting the snapshot, and `load_complete_history` stops when a
round returns the cursor it started from.
---
 .../protocols/v2/observation/messages.lua     |  51 ++-
 lua/opencode/protocols/v2/types.lua           |   8 +-
 .../protocol_v2_observation_runtime_spec.lua  | 323 ++++++++++++++++++
 3 files changed, 369 insertions(+), 13 deletions(-)

diff --git a/lua/opencode/protocols/v2/observation/messages.lua b/lua/opencode/protocols/v2/observation/messages.lua
index 9806e14c..2ac9b251 100644
--- a/lua/opencode/protocols/v2/observation/messages.lua
+++ b/lua/opencode/protocols/v2/observation/messages.lua
@@ -67,6 +67,9 @@ function M.ingest_snapshot(observation, messages, merge)
     end)
   else
     local state = observation:read()
+    -- Only a wholesale replacement moves entries, order and cursor together,
+    -- and only that makes the pages cut from the previous one stale.
+    observation._v2_snapshot_generation = observation._v2_snapshot_generation + 1
     local entries_by_id, order = {}, {}
     for _, entry in ipairs(mapped) do
       entries_by_id[entry.id] = entries.replace(state.entries_by_id[entry.id], entry)
@@ -529,7 +532,8 @@ function M.initialize(observation)
   observation._v2_content_by_message = {}
   observation._v2_older_cursor = nil
   observation._v2_history_complete = false
-  observation._v2_older_loading = false
+  observation._v2_older_page = nil
+  observation._v2_snapshot_generation = 0
 end
 
 ---@param observation OpencodeV2Observation
@@ -537,21 +541,29 @@ end
 function M.attach_history(observation, connection)
   ---@return Promise<nil>
   function observation:load_older()
-    if self._v2_older_loading then
-      boundary.fail('load_older is already in progress')
-    end
     if self._v2_history_complete or not self._v2_older_cursor then
       return Promise.new():resolve(nil)
     end
 
-    self._v2_older_loading = true
+    local in_flight = self._v2_older_page
+    if in_flight then
+      if in_flight.cursor == self._v2_older_cursor then
+        return in_flight.promise
+      end
+      -- The cursor moved while the page was in flight; queue behind it so this
+      -- caller reads the cursor that a rebuild left behind, not the one it saw.
+      return Promise.async(function()
+        in_flight.promise:await()
+        self:load_older():await()
+      end)()
+    end
+
     local finish = self:_begin_local_operation()
     local cursor = self._v2_older_cursor
-    local revision = self._event_revisions.messages
+    local generation = self._v2_snapshot_generation
     local ok, operation_request =
       pcall(connection.operations.list_messages, connection, self._session_id, cursor, 50, nil)
     if not ok then
-      self._v2_older_loading = false
       finish()
       error(operation_request, 0)
     end
@@ -560,25 +572,38 @@ function M.attach_history(observation, connection)
       if not self:_is_current() then
         boundary.fail('older messages arrived after Observation release')
       end
-      if self._event_revisions.messages ~= revision then
-        self:read().sync.messages = { state = 'stale' }
-        self:_start_resource('messages')
+      if self._v2_snapshot_generation ~= generation then
+        -- The snapshot this page was cut from is gone; the newer one already
+        -- carries the server's answer.
         return
       end
       M.apply_page(self, page, true)
       self:_notify('messages')
     end)
-    return request:finally(function()
-      self._v2_older_loading = false
+    local page = { cursor = cursor, promise = nil }
+    ---@cast page OpencodeV2OlderPage
+    page.promise = request:finally(function()
+      -- release/initialize drop the slot; a page landing after that must not
+      -- blank the one that replaced it
+      if self._v2_older_page == page then
+        self._v2_older_page = nil
+      end
       finish()
     end)
+    self._v2_older_page = page
+    return page.promise
   end
 
   ---@return Promise<nil>
   function observation:load_complete_history()
     return Promise.async(function()
       while not self._v2_history_complete and self._v2_older_cursor do
+        local cursor = self._v2_older_cursor
         self:load_older():await()
+        if self._v2_older_cursor == cursor then
+          -- Same cursor back means another round would ask the same question.
+          return
+        end
       end
     end)()
   end
@@ -595,6 +620,8 @@ function M.release(observation)
   observation._v2_content_by_message = {}
   observation._v2_older_cursor = nil
   observation._v2_history_complete = false
+  observation._v2_older_page = nil
+  observation._v2_snapshot_generation = 0
 end
 
 return M
diff --git a/lua/opencode/protocols/v2/types.lua b/lua/opencode/protocols/v2/types.lua
index 7b05102b..d89961fe 100644
--- a/lua/opencode/protocols/v2/types.lua
+++ b/lua/opencode/protocols/v2/types.lua
@@ -301,6 +301,11 @@
 
 ---@alias OpencodeV2EventHandler fun(observation: OpencodeV2Observation, event: OpencodeV2Event, data: table): false|'terminal'|nil
 
+---An older page being fetched; callers asking for the same page share `promise`.
+---@class OpencodeV2OlderPage
+---@field cursor string
+---@field promise Promise<nil>
+
 ---@class OpencodeV2Observation: OpencodeObservation
 ---@field _connection OpencodeV2Connection
 ---@field _v2_admissions table<string, OpencodeV2PendingAdmission|nil>
@@ -312,7 +317,8 @@
 ---@field _v2_content_by_message table<string, table<string, table|nil>|nil>
 ---@field _v2_older_cursor? string
 ---@field _v2_history_complete boolean
----@field _v2_older_loading boolean
+---@field _v2_older_page? OpencodeV2OlderPage
+---@field _v2_snapshot_generation integer Bumped when a wholesale snapshot replaces entries, order and cursor
 ---@field _v2_inbox_terminal table<string, table|nil>
 ---@field _v2_permission_terminal table<string, {answer: string}|nil>
 ---@field _v2_question_terminal table<string, {status: string, answers?: OpencodeV2FormAnswers}|nil>
diff --git a/tests/unit/protocol_v2_observation_runtime_spec.lua b/tests/unit/protocol_v2_observation_runtime_spec.lua
index d19f3eb2..5e0bf512 100644
--- a/tests/unit/protocol_v2_observation_runtime_spec.lua
+++ b/tests/unit/protocol_v2_observation_runtime_spec.lua
@@ -425,6 +425,329 @@ describe('V2 protocol Observation runtime', function()
     stop()
   end)
 
+  it('gives a second caller the page already in flight instead of failing it', function()
+    local value = connection()
+    local calls = {}
+    local older = Promise.new()
+    install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if cursor == nil then
+          return resolved({ data = { user('B', 'B', 4) }, cursor = { next = 'older-cursor' } })
+        end
+        return older
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    local first = observed:load_older()
+    local second = observed:load_older()
+    assert.is_true(first == second, 'both callers wait on the same request')
+    assert.equals(2, #calls)
+
+    older:resolve({ data = { user('A', 'A', 3) }, cursor = {} })
+    first:wait()
+    second:wait()
+    assert.same({ 'A', 'B' }, observed:read().entry_order)
+    assert.equals(2, #calls)
+    stop()
+  end)
+
+  it('fetches the cursor a rebuild produced only after the in-flight page lands', function()
+    local value = connection()
+    local calls = {}
+    local older = Promise.new()
+    local snapshot_cursor = 'c1'
+    local streams = install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if cursor == nil then
+          return resolved({ data = { user('B', 'B', 4) }, cursor = { next = snapshot_cursor } })
+        end
+        if cursor == 'c1' then
+          return older
+        end
+        return resolved({ data = {}, cursor = {} })
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    local first = observed:load_older()
+    assert.equals(2, #calls)
+
+    snapshot_cursor = 'c2'
+    emit(streams[1], event('ses-main', 'session.compaction.ended', {}))
+    flush(function()
+      return observed._v2_older_cursor == 'c2'
+    end)
+    assert.is_false(first:is_resolved(), 'the page in flight is still pending')
+
+    local queued = observed:load_older()
+    local before = #calls
+    for index = 1, before do
+      assert.are_not.equals('c2', calls[index].cursor, 'the new cursor must wait for the page in flight')
+    end
+
+    older:resolve({ data = { user('A', 'A', 3) }, cursor = { next = 'c1b' } })
+    flush(function()
+      for index = before + 1, #calls do
+        if calls[index].cursor == 'c2' then
+          return true
+        end
+      end
+      return false
+    end)
+    queued:wait()
+    -- The rebuild replaced entries, order and cursor, so the page the caller
+    -- queued behind no longer describes this history.
+    assert.is_nil(observed:read().entries_by_id['A'])
+    assert.equals('c2', calls[#calls].cursor)
+    stop()
+  end)
+
+  it('clears the in-flight page when the request fails so the next call starts over', function()
+    local value = connection()
+    local calls = {}
+    local older = Promise.new()
+    local pending = true
+    local streams = install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if cursor == nil then
+          return resolved({ data = { user('B', 'B', 4) }, cursor = { next = 'c1' } })
+        end
+        if cursor == 'c1' and pending then
+          pending = false
+          return older
+        end
+        return resolved({ data = { user('C', 'C', 2) }, cursor = { next = 'c1b' } })
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    local first = observed:load_older()
+    local waited = observed:load_older()
+    assert.equals(2, #calls, 'the second caller shares the request in flight')
+
+    older:reject('HTTP request cancelled')
+    flush(function()
+      return observed._v2_older_page == nil
+    end)
+    for _, promise in ipairs({ first, waited }) do
+      local ok, err = pcall(promise.wait, promise)
+      assert.is_false(ok, 'a cancelled page rejects its waiters')
+      assert.truthy(tostring(err):find('cancelled'), tostring(err))
+    end
+
+    local again = observed:load_older()
+    assert.equals(3, #calls, 'a failed page must not strand the slot')
+    assert.equals('c1', calls[3].cursor, 'the cursor did not move, so the page is asked for again')
+    again:wait()
+    assert.is_not_nil(observed:read().entries_by_id['C'], 'the retried page lands')
+    stop()
+  end)
+
+  it('issues one page for twenty rapid load_older calls', function()
+    local value = connection()
+    local calls = {}
+    local older = Promise.new()
+    local streams = install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if cursor == nil then
+          return resolved({ data = { user('B', 'B', 4) }, cursor = { next = 'c1' } })
+        end
+        if cursor == 'c1' then
+          return older
+        end
+        return resolved({ data = {}, cursor = {} })
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    local pages = {}
+    for _ = 1, 20 do
+      pages[#pages + 1] = observed:load_older()
+    end
+    assert.equals(2, #calls, 'twenty key repeats share the one page in flight')
+
+    older:resolve({ data = { user('A', 'A', 3) }, cursor = { next = 'c1b' } })
+    for _, page in ipairs(pages) do
+      page:wait()
+    end
+    flush(function()
+      return observed:read().entries_by_id['A'] ~= nil
+    end)
+    assert.equals(2, #calls, 'the repeats never started a second request')
+    assert.is_not_nil(observed:read().entries_by_id['A'])
+    stop()
+  end)
+
+  it('applies an older page that arrived after live deltas and keeps the newer copy of shared entries', function()
+    local value = connection()
+    local calls = {}
+    local older = Promise.new()
+    local streams = install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if cursor == nil then
+          return resolved({ data = { user('B', 'B', 4) }, cursor = { next = 'c1' } })
+        end
+        if cursor == 'c1' then
+          return older
+        end
+        return resolved({ data = {}, cursor = {} })
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    local page = observed:load_older()
+    assert.equals(2, #calls)
+    emit(
+      streams[1],
+      event('ses-main', 'session.step.started', { assistantMessageID = 'msg-tail', agent = 'build' }, 20)
+    )
+    emit(streams[1], event('ses-main', 'session.text.started', { assistantMessageID = 'msg-tail', ordinal = 0 }, 21))
+    emit(
+      streams[1],
+      event('ses-main', 'session.text.delta', { assistantMessageID = 'msg-tail', ordinal = 0, delta = 'live' }, 22)
+    )
+    flush()
+
+    local online = observed:read().entries_by_id['B']
+    older:resolve({ data = { user('B', 'stale copy', 4), user('A', 'A', 3) }, cursor = { next = 'c1b' } })
+    page:wait()
+
+    assert.same({ 'A', 'B', 'msg-tail' }, observed:read().entry_order, 'the page prepends without repeating B')
+    assert.equals(online, observed:read().entries_by_id['B'], 'the entry received online keeps its identity')
+    assert.equals('B', observed:read().entries_by_id['B'].content[1].text, 'live content is not replaced by the page')
+    assert.equals('c1b', observed._v2_older_cursor)
+    assert.equals(2, #calls, 'the deltas did not cost another request')
+    stop()
+  end)
+
+  it('drops an older page a fresh snapshot superseded and does not resurrect what it removed', function()
+    local value = connection()
+    local calls = {}
+    local older = Promise.new()
+    local streams = install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if cursor == nil then
+          if #calls == 1 then
+            return resolved({ data = { user('B', 'B', 4) }, cursor = { next = 'c1' } })
+          end
+          return resolved({ data = { user('A', 'A', 3) }, cursor = { next = 'c2' } })
+        end
+        if cursor == 'c1' then
+          return older
+        end
+        return resolved({ data = {}, cursor = {} })
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    local page = observed:load_older()
+    assert.equals(2, #calls)
+    emit(streams[1], event('ses-main', 'session.revert.committed', { messageID = 'B' }, 20))
+    flush(function()
+      return observed._loading.messages == nil and observed:read().sync.messages.state == 'current' and #calls >= 3
+    end)
+    local refetched = #calls
+
+    older:resolve({ data = { user('B', 'B', 4), user('Z', 'Z', 1) }, cursor = {} })
+    page:wait()
+
+    assert.same({ 'A' }, observed:read().entry_order, 'the superseded page is dropped, not merged')
+    assert.equals('c2', observed._v2_older_cursor, 'the cursor stays with the snapshot that replaced the page')
+    assert.equals(refetched, #calls, 'dropping the page does not trigger another snapshot request')
+    stop()
+  end)
+
+  it('stops load_complete_history when a round makes no progress', function()
+    local value = connection()
+    local calls = {}
+    install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if cursor == nil then
+          return resolved({ data = { user('B', 'B', 4) }, cursor = { next = 'c1' } })
+        end
+        return resolved({ data = { user('A', 'A', 3) }, cursor = { next = cursor } })
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    observed:load_complete_history():wait()
+    assert.equals(2, #calls, 'the cursor came back unchanged, so the walk stops after one round')
+    assert.is_not_nil(observed:read().entries_by_id['A'])
+    stop()
+  end)
+
+  it('discards a rebuild that a later event superseded', function()
+    local value = connection()
+    local pulls = 0
+    local rebuild = Promise.new()
+    local streams = install_operations(value, {
+      list_messages = function()
+        pulls = pulls + 1
+        if pulls == 1 then
+          return resolved({ data = { user('B', 'B', 4) }, cursor = { next = 'c1' } })
+        end
+        if pulls == 2 then
+          return rebuild
+        end
+        return resolved({ data = { user('C', 'C', 5) }, cursor = { next = 'c3' } })
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    emit(streams[1], event('ses-main', 'session.compaction.ended', {}, 20))
+    flush(function()
+      return pulls == 2
+    end)
+    rebuild:resolve({ data = { user('A', 'A', 3) }, cursor = { next = 'c2' } })
+    flush(function()
+      return observed._loading.messages == nil and observed:read().entries_by_id['C'] ~= nil
+    end)
+
+    assert.same({ 'C' }, observed:read().entry_order, 'the superseded rebuild never lands')
+    assert.equals('c3', observed._v2_older_cursor)
+    stop()
+  end)
+
   it('keeps terminal inbox and form facts when a stale pending snapshot arrives', function()
     local value = connection()
     local inbox_page, form_page, permission_page = Promise.new(), Promise.new(), Promise.new()

From 040d26fb63de9abd80aee1586b6c4763260454b7 Mon Sep 17 00:00:00 2001
From: jensenojs <jensenojs@qq.com>
Date: Sat, 24 Oct 2026 22:42:19 +0800
Subject: [PATCH 2/3] feat(nav): reach user messages the window has not loaded

`[u`/`]u` searched rendered messages only, so a target above the loaded window
reported `No previous user message`. When the search misses and the window is
still pulling history, the jump is registered and runs after the pull lands.

The position is held by message id, because growing the window renumbers the
lines below it.
---
 lua/opencode/ui/navigation.lua              | 78 +++++++++++++--------
 lua/opencode/ui/renderer.lua                | 31 +++++++-
 lua/opencode/ui/renderer/ctx.lua            |  2 +
 tests/unit/navigation_user_message_spec.lua | 72 +++++++++++++++++++
 4 files changed, 153 insertions(+), 30 deletions(-)

diff --git a/lua/opencode/ui/navigation.lua b/lua/opencode/ui/navigation.lua
index e3c97a39..d29e1ff7 100644
--- a/lua/opencode/ui/navigation.lua
+++ b/lua/opencode/ui/navigation.lua
@@ -94,7 +94,27 @@ function M.goto_prev_message()
   vim.api.nvim_win_set_cursor(win, { 1, 0 })
 end
 
-function M.goto_next_user_message()
+---@param win integer
+---@param direction 'next'|'prev'
+---@param line integer Line the search starts from, 1-indexed
+---@return boolean moved
+local function jump_to_user_message(win, direction, line)
+  local target
+  if direction == 'next' then
+    target = renderer.get_next_user_message(line)
+  else
+    target = renderer.get_prev_user_message(line)
+  end
+  if not target or not target.line_start then
+    return false
+  end
+  mark_jump_position(win)
+  vim.api.nvim_win_set_cursor(win, { target.line_start + 1, 0 })
+  return true
+end
+
+---@param direction 'next'|'prev'
+local function goto_user_message(direction)
   require('opencode.ui.ui').focus_output()
   local windows = state.windows or {}
   local win = windows.output_win
@@ -104,42 +124,44 @@ function M.goto_next_user_message()
     return
   end
 
-  -- Like `gg`, under lazy render the target
-  -- message may not yet have a line_start, so force a full render first.
-  renderer.load_all_messages(nil, { scroll_to_top = false })
+  local missing = direction == 'next' and 'No next user message' or 'No previous user message'
+  -- Growing the window renumbers every line below, so hold the position by id.
+  local anchor = renderer.get_prev_rendered_message(vim.api.nvim_win_get_cursor(win)[1] + 1)
+  local anchor_id = anchor and anchor.message.id
+  local _, started = renderer.load_all_messages(nil, { scroll_to_top = false })
 
-  local current_line = vim.api.nvim_win_get_cursor(win)[1]
-  local next_message = renderer.get_next_user_message(current_line)
-  if next_message and next_message.line_start then
-    mark_jump_position(win)
-    vim.api.nvim_win_set_cursor(win, { next_message.line_start + 1, 0 })
+  if jump_to_user_message(win, direction, vim.api.nvim_win_get_cursor(win)[1]) then
     return
   end
 
-  vim.notify('No next user message', vim.log.levels.INFO)
-end
-
-function M.goto_prev_user_message()
-  require('opencode.ui.ui').focus_output()
-  local windows = state.windows or {}
-  local win = windows.output_win
-  local buf = windows.output_buf
-
-  if not win or not buf then
+  if not started then
+    vim.notify(missing, vim.log.levels.INFO)
     return
   end
 
-  renderer.load_all_messages(nil, { scroll_to_top = false })
+  renderer.after_history_load(function()
+    if not vim.api.nvim_win_is_valid(win) then
+      return
+    end
+    local rendered = anchor_id and renderer.get_rendered_message(anchor_id)
+    local line
+    if rendered and rendered.line_start then
+      line = rendered.line_start + 2
+    else
+      line = vim.api.nvim_win_get_cursor(win)[1]
+    end
+    if not jump_to_user_message(win, direction, line) then
+      vim.notify(missing, vim.log.levels.INFO)
+    end
+  end)
+end
 
-  local current_line = vim.api.nvim_win_get_cursor(win)[1]
-  local previous_message = renderer.get_prev_user_message(current_line)
-  if previous_message and previous_message.line_start then
-    mark_jump_position(win)
-    vim.api.nvim_win_set_cursor(win, { previous_message.line_start + 1, 0 })
-    return
-  end
+function M.goto_next_user_message()
+  goto_user_message('next')
+end
 
-  vim.notify('No previous user message', vim.log.levels.INFO)
+function M.goto_prev_user_message()
+  goto_user_message('prev')
 end
 
 ---@param raw string
diff --git a/lua/opencode/ui/renderer.lua b/lua/opencode/ui/renderer.lua
index fd0ab584..649bb268 100644
--- a/lua/opencode/ui/renderer.lua
+++ b/lua/opencode/ui/renderer.lua
@@ -836,6 +836,16 @@ local function grow_window_with_older_page(ctx)
   return true
 end
 
+---Drop the pending navigation callback once the scroll target it was registered
+---under is gone.
+---@param ctx RendererCtx
+local function drop_stale_callback(ctx)
+  local pending = ctx.after_history_load
+  if pending and pending.revision ~= ctx.history_navigation_revision then
+    ctx.after_history_load = nil
+  end
+end
+
 ---Pull the complete remaining history and render all of it.
 ---@return boolean Whether a history load was started
 ---@param ctx RendererCtx
@@ -856,6 +866,7 @@ local function load_complete_history(ctx, scroll_to_top)
   end
   request
     :and_then(function()
+      drop_stale_callback(ctx)
       if
         not ctx:is_active()
         or ctx.observation ~= observation
@@ -872,8 +883,14 @@ local function load_complete_history(ctx, scroll_to_top)
         pcall(output_window.restore_view_topline, win, 1)
       end
       ctx.history_top_pending = false
+      local after_load = ctx.after_history_load
+      ctx.after_history_load = nil
+      if after_load then
+        after_load.fn()
+      end
     end)
     :catch(function(err)
+      drop_stale_callback(ctx)
       if ctx.history_navigation_revision == navigation_revision then
         ctx.history_top_pending = false
       end
@@ -1043,7 +1060,8 @@ end
 
 ---Load all remaining messages and re-render.
 ---Used for navigation and search; by default, gg lands at the true top.
----@return boolean Whether any messages were loaded
+---@return boolean loaded Whether any messages were loaded
+---@return boolean started Whether a complete-history pull was started
 ---@param ctx? RendererCtx
 ---@param opts? {scroll_to_top?: boolean}
 function M.load_all_messages(ctx, opts)
@@ -1069,7 +1087,16 @@ function M.load_all_messages(ctx, opts)
   if not loading then
     ctx.history_top_pending = false
   end
-  return loading or expanded
+  return loading or expanded, loading
+end
+
+---Run `fn` when the complete-history pull in flight lands; a later `gg` drops it
+---instead of running it under a different scroll target.
+---@param fn fun()
+---@param ctx? RendererCtx
+function M.after_history_load(fn, ctx)
+  ctx = ctx or contexts.current()
+  ctx.after_history_load = { fn = fn, revision = ctx.history_navigation_revision }
 end
 
 ---Render the currently observed state synchronously; this does not load history.
diff --git a/lua/opencode/ui/renderer/ctx.lua b/lua/opencode/ui/renderer/ctx.lua
index b51d7fe8..5f3c79ec 100644
--- a/lua/opencode/ui/renderer/ctx.lua
+++ b/lua/opencode/ui/renderer/ctx.lua
@@ -35,6 +35,7 @@ M.prompt_controllers = {}
 ---@field history_expanded boolean
 ---@field history_navigation_revision integer
 ---@field history_top_pending boolean
+---@field after_history_load {fn: fun(), revision: integer}|nil
 ---@field get_child_parts fun(session_id: string): table[]|nil
 ---@field prompt_controllers {permission?: PermissionController, question?: QuestionController}
 ---@field formatted_parts table<string, Output>
@@ -140,6 +141,7 @@ function ctx:reset()
   self.history_expanded = false
   self.history_navigation_revision = 0
   self.history_top_pending = false
+  self.after_history_load = nil
   self.file_revision = 0
   self.needs_reconcile = false
   self:bulk_reset()
diff --git a/tests/unit/navigation_user_message_spec.lua b/tests/unit/navigation_user_message_spec.lua
index cc475b84..ce7c6cc9 100644
--- a/tests/unit/navigation_user_message_spec.lua
+++ b/tests/unit/navigation_user_message_spec.lua
@@ -299,6 +299,78 @@ describe('navigation user message jumps', function()
 
       assert.equals(2, vim.api.nvim_win_get_cursor(output_win)[1])
     end)
+
+    it('jumps once the pulled history renders the target above the window', function()
+      seed({
+        { id = 'a1', kind = 'assistant', line_start = 1 },
+      })
+      contexts.current().lazy_render_count = 1
+
+      local pending
+      local original_load = renderer.load_all_messages
+      local original_after = renderer.after_history_load
+      renderer.load_all_messages = function()
+        return true, true
+      end
+      renderer.after_history_load = function(fn)
+        pending = fn
+      end
+      local notify_stub = stub(vim, 'notify')
+
+      vim.api.nvim_win_set_cursor(output_win, { 2, 0 })
+      navigation.goto_prev_user_message()
+
+      renderer.load_all_messages = original_load
+      renderer.after_history_load = original_after
+      notify_stub:revert()
+
+      assert.is_function(pending)
+      assert.stub(notify_stub).was_called(0)
+
+      -- the pull lands: an older user message joins above and everything moves down
+      local ctx = contexts.current()
+      local older = { id = 'u0', kind = 'user', content = {} }
+      table.insert(ctx.entries, 1, older)
+      ctx.render_state:set_message(older, 10, 10)
+      ctx.render_state:set_message(ctx.entries[2], 30, 30)
+      pending()
+
+      assert.equals(11, vim.api.nvim_win_get_cursor(output_win)[1])
+    end)
+
+    it('jumps forward once the pulled history renders the target below the window', function()
+      seed({
+        { id = 'u1', kind = 'user', line_start = 1 },
+        { id = 'u2', kind = 'user' },
+      })
+      contexts.current().lazy_render_count = 1
+
+      local pending
+      local original_load = renderer.load_all_messages
+      local original_after = renderer.after_history_load
+      renderer.load_all_messages = function()
+        return true, true
+      end
+      renderer.after_history_load = function(fn)
+        pending = fn
+      end
+      local notify_stub = stub(vim, 'notify')
+
+      vim.api.nvim_win_set_cursor(output_win, { 2, 0 })
+      navigation.goto_next_user_message()
+
+      renderer.load_all_messages = original_load
+      renderer.after_history_load = original_after
+      notify_stub:revert()
+
+      assert.is_function(pending)
+      assert.stub(notify_stub).was_called(0)
+
+      contexts.current().render_state:set_message(contexts.current().entries[2], 10, 10)
+      pending()
+
+      assert.equals(11, vim.api.nvim_win_get_cursor(output_win)[1])
+    end)
   end)
 
   it('keeps user-message jumps after a completed history request resolves asynchronously', function()

From 95c9de53f817e9dffa96c0c7a2d574578db5b8dc Mon Sep 17 00:00:00 2001
From: jensenojs <jensenojs@qq.com>
Date: Sat, 24 Oct 2026 22:42:19 +0800
Subject: [PATCH 3/3] fix(v2): update the transcript in place instead of
 rebuilding it

Three events refetched the whole transcript, which replaced every loaded page
with the newest one.

A background completion now projects the entry the server persists under the
event's id. A committed revert drops the tail it deleted, since upstream deletes
from the boundary onwards. A compaction still refetches, because the server's
history starts at that compaction.

A handler that does refetch only marks the resource stale; the dispatch starts
the request after bumping the revision, so one event costs one request.
---
 lua/opencode/protocols/v2/observation.lua     |   2 +-
 .../protocols/v2/observation/messages.lua     |  61 ++++++--
 .../protocol_v2_observation_runtime_spec.lua  |  68 ++++++++-
 tests/unit/protocol_v2_observation_spec.lua   |  72 +++++++--
 tests/unit/renderer_lazy_spec.lua             | 138 ++++++++++++++++++
 5 files changed, 317 insertions(+), 24 deletions(-)

diff --git a/lua/opencode/protocols/v2/observation.lua b/lua/opencode/protocols/v2/observation.lua
index 28303df0..43cf5cc4 100644
--- a/lua/opencode/protocols/v2/observation.lua
+++ b/lua/opencode/protocols/v2/observation.lua
@@ -30,7 +30,7 @@ function M.new(connection, ref)
     ---@param sync table
     ---@return boolean
     refresh_after_event = function(resource, sync)
-      return resource ~= 'files' and sync.state == 'error'
+      return resource ~= 'files' and (sync.state == 'error' or sync.state == 'stale')
     end,
     request_resource = function(current, resource)
       ---@cast current OpencodeV2Observation
diff --git a/lua/opencode/protocols/v2/observation/messages.lua b/lua/opencode/protocols/v2/observation/messages.lua
index 2ac9b251..b81db5d3 100644
--- a/lua/opencode/protocols/v2/observation/messages.lua
+++ b/lua/opencode/protocols/v2/observation/messages.lua
@@ -250,26 +250,30 @@ end
 ---@param data table
 handlers['session.synthetic'] = function(observation, event, data)
   local metadata = data.metadata
-  -- Background completions carry no message id, so the snapshot is the only
-  -- source of the entry that records their outcome.
-  if normalize.is_background_notification(metadata) then
-    return invalidate_messages()
-  end
-  -- Only server-authored notices carry a description; plain synthetic parts are
-  -- already in the message snapshot, so projecting them again would duplicate.
-  if type(data.description) ~= 'string' or data.description == '' then
-    return false
+  local background = normalize.is_background_notification(metadata)
+  -- Pulling a snapshot for these would replace every loaded page with the
+  -- newest one, and the entry is already reachable from the event's id.
+  if not background then
+    -- Plain synthetic parts are already in the snapshot; only notices carry a
+    -- description meant for the transcript.
+    if type(data.description) ~= 'string' or data.description == '' then
+      return false
+    end
   end
   if type(data.text) ~= 'string' then
     return invalid(observation, event.type, 'is missing text')
   end
-  return put_notice(observation, event, {
+  local info = {
     id = notice_message_id(event),
     type = 'synthetic',
     text = data.text,
     description = data.description,
     time = { created = event.created },
-  })
+  }
+  if background then
+    info.metadata = metadata
+  end
+  return put_notice(observation, event, info)
 end
 
 ---@param observation OpencodeV2Observation
@@ -476,7 +480,37 @@ end
 handlers['session.tool.success'] = finish_tool
 handlers['session.tool.failed'] = finish_tool
 
-handlers['session.revert.committed'] = invalidate_messages
+---A committed revert deletes every message from its boundary onwards, so a
+---loaded tail can be dropped without asking for the page again.
+---@param observation OpencodeV2Observation
+---@param event OpencodeV2Event
+---@param data table
+handlers['session.revert.committed'] = function(observation, event, data)
+  local order = observation:read().entry_order
+  local boundary
+  if type(data.to) == 'string' then
+    for position, id in ipairs(order) do
+      if id == data.to then
+        boundary = position
+        break
+      end
+    end
+  end
+  if not boundary then
+    -- The boundary predates the loaded window, so any loaded message may have
+    -- been deleted; only a snapshot can say which.
+    return invalidate_messages()
+  end
+  local state = observation:read()
+  for position = #order, boundary, -1 do
+    local id = order[position]
+    state.entries_by_id[id] = nil
+    observation._v2_content_by_message[id] = nil
+    order[position] = nil
+  end
+  return true
+end
+
 handlers['session.compaction.ended'] = invalidate_messages
 
 ---@param observation OpencodeV2Observation
@@ -492,8 +526,9 @@ function M.ingest_event(observation, event)
   end
   local result = handler(observation, event, event.data)
   if result == 'refetch' then
+    -- Let `_event_changed` start the refetch: it bumps the revision first, so
+    -- the request it makes is the one that lands.
     observation:read().sync.messages = { state = 'stale' }
-    observation:_start_resource('messages')
     return true
   end
   if not result then
diff --git a/tests/unit/protocol_v2_observation_runtime_spec.lua b/tests/unit/protocol_v2_observation_runtime_spec.lua
index 5e0bf512..143447a0 100644
--- a/tests/unit/protocol_v2_observation_runtime_spec.lua
+++ b/tests/unit/protocol_v2_observation_runtime_spec.lua
@@ -673,11 +673,12 @@ describe('V2 protocol Observation runtime', function()
 
     local page = observed:load_older()
     assert.equals(2, #calls)
-    emit(streams[1], event('ses-main', 'session.revert.committed', { messageID = 'B' }, 20))
+    emit(streams[1], event('ses-main', 'session.compaction.ended', {}, 20))
     flush(function()
       return observed._loading.messages == nil and observed:read().sync.messages.state == 'current' and #calls >= 3
     end)
     local refetched = #calls
+    assert.equals(3, refetched, 'one rebuild request, not a stale one followed by a second')
 
     older:resolve({ data = { user('B', 'B', 4), user('Z', 'Z', 1) }, cursor = {} })
     page:wait()
@@ -688,6 +689,70 @@ describe('V2 protocol Observation runtime', function()
     stop()
   end)
 
+  it('lets a rebuild of an empty window replace text deltas appended while it flew', function()
+    local value = connection()
+    local calls = {}
+    local rebuild = Promise.new()
+    local function reply(text)
+      return { id = 'msg-reply', type = 'assistant', agent = 'build', content = { { type = 'text', text = text } } }
+    end
+    local streams = install_operations(value, {
+      list_messages = function(_, _, cursor, limit)
+        calls[#calls + 1] = { cursor = cursor, limit = limit }
+        if #calls == 1 then
+          return resolved({ data = {}, cursor = { next = 'c1' } })
+        end
+        return rebuild
+      end,
+    })
+    local observed = value:observe({ id = 'ses-main' })
+    local stop = observed:watch({ 'messages' }, function() end)
+    flush(function()
+      return observed:read().sync.messages.state == 'current'
+    end)
+
+    emit(
+      streams[1],
+      event('ses-main', 'session.step.started', { assistantMessageID = 'msg-reply', agent = 'build' }, 20)
+    )
+    emit(streams[1], event('ses-main', 'session.text.started', { assistantMessageID = 'msg-reply', ordinal = 0 }, 21))
+    emit(
+      streams[1],
+      event('ses-main', 'session.text.delta', { assistantMessageID = 'msg-reply', ordinal = 0, delta = 'b' }, 22)
+    )
+    emit(
+      streams[1],
+      event('ses-main', 'session.text.delta', { assistantMessageID = 'msg-reply', ordinal = 0, delta = 'c' }, 23)
+    )
+    flush()
+    assert.equals('bc', observed:read().entries_by_id['msg-reply'].content[1].text, 'the deltas append locally')
+
+    emit(streams[1], event('ses-main', 'session.revert.committed', { messageID = 'msg-reply' }, 24))
+    flush(function()
+      return #calls >= 2
+    end)
+    rebuild:resolve({ data = { reply('b') }, cursor = { next = 'c1' } })
+    flush(function()
+      return observed._loading.messages == nil and observed:read().sync.messages.state == 'current'
+    end)
+
+    assert.equals(
+      'b',
+      observed:read().entries_by_id['msg-reply'].content[1].text,
+      'the snapshot is authoritative even though it lags the live text'
+    )
+
+    emit(
+      streams[1],
+      event('ses-main', 'session.text.ended', { assistantMessageID = 'msg-reply', ordinal = 0, text = 'bc' }, 25)
+    )
+    flush(function()
+      return observed:read().entries_by_id['msg-reply'].content[1].text == 'bc'
+    end)
+    assert.equals('bc', observed:read().entries_by_id['msg-reply'].content[1].text, 'the finished text restores it')
+    stop()
+  end)
+
   it('stops load_complete_history when a round makes no progress', function()
     local value = connection()
     local calls = {}
@@ -738,6 +803,7 @@ describe('V2 protocol Observation runtime', function()
     flush(function()
       return pulls == 2
     end)
+    emit(streams[1], event('ses-main', 'session.compaction.ended', {}, 21))
     rebuild:resolve({ data = { user('A', 'A', 3) }, cursor = { next = 'c2' } })
     flush(function()
       return observed._loading.messages == nil and observed:read().entries_by_id['C'] ~= nil
diff --git a/tests/unit/protocol_v2_observation_spec.lua b/tests/unit/protocol_v2_observation_spec.lua
index da9a352e..d79edd8e 100644
--- a/tests/unit/protocol_v2_observation_spec.lua
+++ b/tests/unit/protocol_v2_observation_spec.lua
@@ -115,7 +115,7 @@ describe('V2 protocol Observation interpretation', function()
     observed._start_resource = original_refresh
   end)
 
-  it('refreshes messages when a background completion notification arrives', function()
+  it('projects a background completion from the event without pulling a snapshot', function()
     local observed = observation('ses-bg')
     local original = observed._start_resource
     local refreshed
@@ -123,16 +123,70 @@ describe('V2 protocol Observation interpretation', function()
       refreshed = resource
     end
 
-    assert.is_true(observation_module.ingest_event(
-      observed,
-      event('ses-bg', 'session.synthetic', {
-        text = '<shell id="sh_1" state="completed" command="make build">\nexit 0\n</shell>',
-        metadata = { source = 'shell', shellID = 'sh_1', state = 'completed', exit = 0 },
-      })
-    ))
-    assert.equals('messages', refreshed)
+    local completion = event('ses-bg', 'session.synthetic', {
+      text = '<shell id="sh_1" state="completed" command="make build">\nexit 0\n</shell>',
+      metadata = { source = 'shell', shellID = 'sh_1', state = 'completed', exit = 0 },
+    })
+    assert.is_true(observation_module.ingest_event(observed, completion))
+
+    local state = observed:read()
+    assert.equals('current', state.sync.messages.state, 'projecting the completion needs no snapshot')
+    assert.equals(1, #state.entry_order)
+    local entry = state.entries_by_id[state.entry_order[1]]
+    assert.equals('sh_1', entry.background_result.id)
+    assert.equals('exit 0', entry.background_result.text)
+
+    observed:_event_changed('messages')
+    assert.is_nil(refreshed, 'the dispatch has nothing to refetch')
+
+    assert.is_true(observation_module.ingest_event(observed, completion))
+    assert.equals(1, #observed:read().entry_order, 'a redelivered completion replaces its entry')
+
+    observed._start_resource = original
+  end)
+
+  it('drops the tail a committed revert deleted without refetching', function()
+    local observed = observation('ses-revert')
+    observation_module.ingest_snapshot(observed, { assistant('msg-c'), assistant('msg-b'), assistant('msg-a') })
+    local original = observed._start_resource
+    local refreshed
+    observed._start_resource = function(_, resource)
+      refreshed = resource
+    end
+
+    assert.is_true(
+      observation_module.ingest_event(observed, event('ses-revert', 'session.revert.committed', { to = 'msg-b' }))
+    )
+
+    local state = observed:read()
+    assert.same({ 'msg-a' }, state.entry_order, 'the boundary message is deleted too')
+    assert.is_nil(state.entries_by_id['msg-c'])
+    assert.equals('current', state.sync.messages.state)
+    assert.is_nil(refreshed, 'a known tail needs no snapshot')
+
+    observed._start_resource = original
+  end)
+
+  it('asks for the snapshot when the revert boundary was never loaded', function()
+    local observed = observation('ses-revert')
+    observation_module.ingest_snapshot(observed, { assistant('msg-a') })
+    local original = observed._start_resource
+    local refreshed
+    observed._start_resource = function(_, resource)
+      refreshed = resource
+    end
+
+    assert.is_true(
+      observation_module.ingest_event(
+        observed,
+        event('ses-revert', 'session.revert.committed', { to = 'msg-before-the-window' })
+      )
+    )
     assert.equals('stale', observed:read().sync.messages.state)
 
+    observed:_event_changed('messages')
+    assert.equals('messages', refreshed, 'only a snapshot can say what the deletion took')
+
     observed._start_resource = original
   end)
 
diff --git a/tests/unit/renderer_lazy_spec.lua b/tests/unit/renderer_lazy_spec.lua
index db679092..42a315e0 100644
--- a/tests/unit/renderer_lazy_spec.lua
+++ b/tests/unit/renderer_lazy_spec.lua
@@ -363,6 +363,41 @@ describe('lazy render', function()
     end
   end)
 
+  it('replaces the transcript with what a rebuilt snapshot carries', function()
+    local session = { id = 'ses_test', location = { directory = helpers.MOCK_CWD } }
+    local entries = make_session_data(50) -- 100 messages, oldest first
+
+    ---@param list table[]
+    local function observed(list)
+      local snapshot = { entry_order = {}, entries_by_id = {}, session = session, sync = {} }
+      for _, entry in ipairs(list) do
+        snapshot.entry_order[#snapshot.entry_order + 1] = entry.id
+        snapshot.entries_by_id[entry.id] = entry
+      end
+      return {
+        read = function()
+          return snapshot
+        end,
+      }
+    end
+
+    contexts.current().observation = observed(entries)
+    contexts.current().entries = entries
+    contexts.current().lazy_render_count = #entries
+    renderer.render_from_cache()
+    assert.are.equal(#entries, count_rendered_messages())
+
+    -- The snapshot carries the newest page only.
+    contexts.current().observation = observed(vim.list_slice(entries, 51, 100))
+    renderer.render_from_cache()
+
+    assert.are.equal(50, count_rendered_messages())
+    local render_state = contexts.current().render_state
+    assert.is_nil(render_state:get_message('msg_u1'))
+    assert.is_nil(render_state:get_message('msg_a25'))
+    assert.is_not_nil(render_state:get_message('msg_u26'))
+  end)
+
   it('load_more_messages returns false when all messages already rendered', function()
     -- No lazy_render_count means all messages rendered — no unrendered content
     local session_data = make_session_data(50) -- 100 messages total
@@ -741,6 +776,109 @@ describe('older history bridge', function()
     assert.are.equal(1, vim.api.nvim_win_get_cursor(state.windows.output_win)[1])
   end)
 
+  it('drops a navigation callback when a top jump supersedes the pull it was registered for', function()
+    config.ui.output.max_messages = 10
+    local newer = make_session_data(20)
+    local entries = newer
+    local page_request = Promise.new()
+    local top_request = Promise.new()
+    local requests = { page_request, top_request }
+    local observation = {
+      read = function()
+        local by_id, order = {}, {}
+        for _, entry in ipairs(entries) do
+          by_id[entry.id] = entry
+          order[#order + 1] = entry.id
+        end
+        return { session = { id = 'ses_test' }, entries_by_id = by_id, entry_order = order }
+      end,
+      load_complete_history = function()
+        return table.remove(requests, 1) or Promise.new()
+      end,
+    }
+    local ctx = contexts.current()
+    ctx.observation = observation
+    ctx.lazy_render_count = 10
+    renderer._render_full_session_data(newer)
+    while ctx.lazy_render_count < #newer do
+      assert.is_true(renderer.load_more_messages())
+    end
+    local first = renderer.get_next_rendered_message(0)
+    vim.api.nvim_win_set_cursor(state.windows.output_win, { first.line_start + 1, 0 })
+
+    local reached = 0
+    assert.is_true(renderer.load_all_messages(nil, { scroll_to_top = false }))
+    renderer.after_history_load(function()
+      reached = reached + 1
+    end)
+
+    require('opencode.ui.navigation').goto_first_message()
+    page_request:resolve(nil)
+    vim.wait(100, function()
+      return false
+    end)
+    assert.are.equal(0, reached, 'a superseded pull must not fire the callback')
+
+    top_request:resolve(nil)
+    assert.is_true(
+      vim.wait(1000, function()
+        return vim.api.nvim_win_get_cursor(state.windows.output_win)[1] == 1
+      end),
+      'the top jump should still land at the first message'
+    )
+    assert.are.equal(0, reached, 'the top jump must not inherit the callback')
+  end)
+
+  it('keeps a callback registered while an older pull is still in flight', function()
+    config.ui.output.max_messages = 10
+    local newer = make_session_data(20)
+    local entries = newer
+    local failing = Promise.new()
+    local sibling = Promise.new()
+    local requests = { failing, sibling }
+    local observation = {
+      read = function()
+        local by_id, order = {}, {}
+        for _, entry in ipairs(entries) do
+          by_id[entry.id] = entry
+          order[#order + 1] = entry.id
+        end
+        return { session = { id = 'ses_test' }, entries_by_id = by_id, entry_order = order }
+      end,
+      load_complete_history = function()
+        return table.remove(requests, 1) or Promise.new()
+      end,
+    }
+    local ctx = contexts.current()
+    ctx.observation = observation
+    ctx.lazy_render_count = 10
+    renderer._render_full_session_data(newer)
+    while ctx.lazy_render_count < #newer do
+      assert.is_true(renderer.load_more_messages())
+    end
+
+    assert.is_true(renderer.load_all_messages(nil, { scroll_to_top = false }))
+    local reached = 0
+    renderer.after_history_load(function()
+      reached = reached + 1
+    end)
+    assert.is_true(renderer.load_all_messages(nil, { scroll_to_top = false }))
+
+    failing:reject('boom')
+    vim.wait(100, function()
+      return false
+    end)
+    assert.are.equal(0, reached, 'a failed pull must not fire the callback')
+
+    sibling:resolve(nil)
+    assert.is_true(
+      vim.wait(1000, function()
+        return reached == 1
+      end),
+      'the sibling pull should still fire the callback'
+    )
+  end)
+
   it('load_more_messages pulls an older page when the cached window is exhausted', function()
     local observation, older, newer, pages_left = observation_with_older_page()
     contexts.current().observation = observation