refactor(event_manager): throttle event emission

Add new module, ThrottlingEmitter, that collects events and "drains"
them every drain_interval_ms. This allows us to batch events which lets
us process a group of them quickly. That, in turn, helps make sure we
don't overwhelm neovim (and particularly treesitter) with a ton of
events.

Before this change, the behavior we'd see is that the neovim ui could
get slow or hang for several/many seconds if we emitted too many events.
Disabling treesitter seemed to eliminate that behavior but it also
caused the buffers to lose their highlighting which wasn't a good user
experience.

Author: cameronr

lua/opencode/event_manager.lua

@@ -1,4 +1,5 @@
 local state = require('opencode.state')
+local ThrottlingEmitter = require('opencode.throttling_emitter')
 
 --- @class EventInstallationUpdated
 --- @field type "installation.updated"
@@ -113,24 +114,34 @@ local state = require('opencode.state')
 --- | "custom.server_ready"
 --- | "custom.server_stopped"
 --- | "custom.restore_point.created"
+--- | "custom.emit_events.started"
+--- | "custom.emit_events.finished"
 
 --- @class EventManager
 --- @field events table<string, function[]> Event listener registry
 --- @field server_subscription table|nil Subscription to server events
 --- @field is_started boolean Whether the event manager is started
 --- @field captured_events table[] List of captured events for debugging
+--- @field throttling_emitter ThrottlingEmitter Throttle instance for batching events
 local EventManager = {}
 EventManager.__index = EventManager
 
 --- Create a new EventManager instance
 --- @return EventManager
 function EventManager.new()
-  return setmetatable({
+  local self = setmetatable({
     events = {},
     server_subscription = nil,
     is_started = false,
     captured_events = {},
   }, EventManager)
+
+  -- TODO: make drain delay configurable
+  self.throttling_emitter = ThrottlingEmitter.new(function(events)
+    self:_on_drained_events(events)
+  end, 40)
+
+  return self
 end
 
 --- Subscribe to an event with type-safe callbacks using function overloads
@@ -155,6 +166,8 @@ end
 --- @overload fun(self: EventManager, event_name: "custom.server_ready", callback: fun(data: ServerReadyEvent['properties']): nil)
 --- @overload fun(self: EventManager, event_name: "custom.server_stopped", callback: fun(data: ServerStoppedEvent['properties']): nil)
 --- @overload fun(self: EventManager, event_name: "custom.restore_point.created", callback: fun(data: RestorePointCreatedEvent['properties']): nil)
+--- @overload fun(self: EventManager, event_name: "custom.emit_events.started", callback: fun(): nil)
+--- @overload fun(self: EventManager, event_name: "custom.emit_events.finished", callback: fun(): nil)
 --- @param event_name OpencodeEventName The event name to listen for
 --- @param callback function Callback function to execute when event is triggered
 function EventManager:subscribe(event_name, callback)
@@ -186,6 +199,8 @@ end
 --- @overload fun(self: EventManager, event_name: "custom.server_ready", callback: fun(data: ServerReadyEvent['properties']): nil)
 --- @overload fun(self: EventManager, event_name: "custom.server_stopped", callback: fun(data: ServerStoppedEvent['properties']): nil)
 --- @overload fun(self: EventManager, event_name: "custom.restore_point.created", callback: fun(data: RestorePointCreatedEvent['properties']): nil)
+--- @overload fun(self: EventManager, event_name: "custom.emit_events.started", callback: fun(): nil)
+--- @overload fun(self: EventManager, event_name: "custom.emit_events.finished", callback: fun(): nil)
 --- @param event_name OpencodeEventName The event name
 --- @param callback function The callback function to remove
 function EventManager:unsubscribe(event_name, callback)
@@ -202,6 +217,21 @@ function EventManager:unsubscribe(event_name, callback)
   end
 end
 
+---Callaback from ThrottlingEmitter when the events are now
+---ready to be processed
+---@param events any
+function EventManager:_on_drained_events(events)
+  self:emit('custom.emit_events.started', {})
+
+  -- TODO: try collapsing events here
+
+  for _, event in ipairs(events) do
+    self:emit(event.type, event.properties)
+  end
+
+  self:emit('custom.emit_events.finished', {})
+end
+
 --- Emit an event to all subscribers
 --- @param event_name OpencodeEventName The event name
 --- @param data any Data to pass to event listeners
@@ -217,7 +247,6 @@ function EventManager:emit(event_name, data)
     table.insert(self.captured_events, vim.deepcopy(event))
   end
 
-  -- schedule events to allow for similar pieces of state to be updated
   for _, callback in ipairs(listeners) do
     pcall(callback, data)
   end
@@ -267,6 +296,7 @@ function EventManager:stop()
   self.is_started = false
   self:_cleanup_server_subscription()
 
+  self.throttling_emitter:clear()
   self.events = {}
 end
 
@@ -282,9 +312,7 @@ function EventManager:_subscribe_to_server_events(server)
   local api_client = state.api_client
 
   local emitter = function(event)
-    vim.schedule(function()
-      self:emit(event.type, event.properties)
-    end)
+    self.throttling_emitter:enqueue(event)
   end
 
   self.server_subscription = api_client:subscribe_to_events(nil, emitter)

lua/opencode/throttling_emitter.lua

@@ -0,0 +1,67 @@
+local M = {}
+
+--- @class ThrottlingEmitter
+--- @field queue table[] Queue of pending items to be processed
+--- @field drain_scheduled boolean Whether a drain is already scheduled
+--- @field process_fn fun(table): nil Function to process the queue of events
+--- @field drain_interval_ms number Interval between drains in milliseconds
+--- @field enqueue fun(self: ThrottlingEmitter, item: any) Enqueue an item for batch processing
+--- @field clear fun(self: ThrottlingEmitter) Clear the queue and cancel any pending drain
+local ThrottlingEmitter = {}
+ThrottlingEmitter.__index = ThrottlingEmitter
+
+--- Create a new ThrottlingEmitter instance. This emitter collects events and
+--- then drains them every drain_interval_ms milliseconds. This is helpful to
+--- make sure we're not generating so many events that we don't overwhelm
+--- neovim, particularly treesitter.
+--- @param process_fn function Function to call for each item
+--- @param drain_interval_ms number? Interval between drains in milliseconds (default 10)
+--- @return ThrottlingEmitter
+function M.new(process_fn, drain_interval_ms)
+  return setmetatable({
+    queue = {},
+    drain_scheduled = false,
+    process_fn = process_fn,
+    drain_interval_ms = drain_interval_ms or 40,
+  }, ThrottlingEmitter)
+end
+
+--- Enqueue an item for batch processing
+--- @param item any The item to enqueue
+function ThrottlingEmitter:enqueue(item)
+  table.insert(self.queue, item)
+
+  if not self.drain_scheduled then
+    self.drain_scheduled = true
+    vim.defer_fn(function()
+      self:_drain()
+    end, self.drain_interval_ms)
+  end
+end
+
+--- Process all queued items
+function ThrottlingEmitter:_drain()
+  self.drain_scheduled = false
+
+  local items_to_process = self.queue
+  self.queue = {}
+
+  self.process_fn(items_to_process)
+
+  -- double check that items weren't added while processing
+  if #self.queue > 0 and not self.drain_scheduled then
+    self.drain_scheduled = true
+    vim.defer_fn(function()
+      self:_drain()
+    end, self.drain_interval_ms)
+  end
+  -- end)
+end
+
+--- Clear the queue and cancel any pending drain
+function ThrottlingEmitter:clear()
+  self.queue = {}
+  self.drain_scheduled = false
+end
+
+return M

lua/opencode/ui/output_window.lua

@@ -105,27 +105,6 @@ function M.set_lines(lines, start_line, end_line)
   vim.api.nvim_set_option_value('modifiable', false, { buf = windows.output_buf })
 end
 
---- FIXME: remove debugging code
----Set text in a specific line at character positions
----@param line integer The line number (0-indexed)
----@param start_col integer The starting column (0-indexed)
----@param end_col integer The ending column (0-indexed)
----@param text string The text to insert
-function M.set_text(line, start_col, end_col, text)
-  if not M.mounted() then
-    return
-  end
-
-  local windows = state.windows
-  if not windows or not windows.output_buf then
-    return
-  end
-
-  vim.api.nvim_set_option_value('modifiable', true, { buf = windows.output_buf })
-  vim.api.nvim_buf_set_text(windows.output_buf, line, start_col, line, end_col, { text })
-  vim.api.nvim_set_option_value('modifiable', false, { buf = windows.output_buf })
-end
-
 ---Clear output buf extmarks
 ---@param start_line? integer Line to start clearing, defaults 0
 ---@param end_line? integer Line to clear until, defaults to -1

lua/opencode/ui/renderer.lua

@@ -10,7 +10,6 @@ local M = {}
 M._subscriptions = {}
 M._prev_line_count = 0
 M._render_state = RenderState.new()
-M._disable_auto_scroll = false
 M._last_part_formatted = { part_id = nil, formatted_data = nil }
 
 local trigger_on_data_rendered = require('opencode.util').debounce(function()
@@ -37,7 +36,6 @@ end, config.ui.output.rendering.markdown_debounce_ms or 250)
 function M.reset()
   M._prev_line_count = 0
   M._render_state:reset()
-  M._disable_auto_scroll = false
   M._last_part_formatted = { part_id = nil, formatted_data = nil }
 
   output_window.clear()
@@ -73,13 +71,14 @@ function M._setup_event_subscriptions(subscribe)
   state.event_manager[method](state.event_manager, 'session.compacted', M.on_session_compacted)
   state.event_manager[method](state.event_manager, 'session.error', M.on_session_error)
   state.event_manager[method](state.event_manager, 'message.updated', M.on_message_updated)
-  state.event_manager[method](state.event_manager, 'message.part.updated', M.on_part_updated)
   state.event_manager[method](state.event_manager, 'message.removed', M.on_message_removed)
+  state.event_manager[method](state.event_manager, 'message.part.updated', M.on_part_updated)
   state.event_manager[method](state.event_manager, 'message.part.removed', M.on_part_removed)
   state.event_manager[method](state.event_manager, 'permission.updated', M.on_permission_updated)
   state.event_manager[method](state.event_manager, 'permission.replied', M.on_permission_replied)
   state.event_manager[method](state.event_manager, 'file.edited', M.on_file_edited)
   state.event_manager[method](state.event_manager, 'custom.restore_point.created', M.on_restore_points)
+  state.event_manager[method](state.event_manager, 'custom.emit_events.finished', M.on_emit_events_finished)
 
   state[method]('is_opencode_focused', M.on_focus_changed)
 end
@@ -121,9 +120,6 @@ end
 function M._render_full_session_data(session_data)
   M.reset()
 
-  -- disable auto-scroll, makes loading a full session much faster
-  M._disable_auto_scroll = true
-
   local revert_index = nil
 
   -- local event_manager = state.event_manager
@@ -146,9 +142,7 @@ function M._render_full_session_data(session_data)
     M._write_formatted_data(formatter._format_revert_message(state.messages, revert_index))
   end
 
-  -- re-enable Auto-scroll
-  M._disable_auto_scroll = false
-  M._scroll_to_bottom()
+  M.scroll_to_bottom()
 end
 
 ---Render lines as the entire output buffer
@@ -169,17 +163,17 @@ function M.render_output(output_data)
   output_window.set_lines(output_data.lines)
   output_window.clear_extmarks()
   output_window.set_extmarks(output_data.extmarks)
-  M._scroll_to_bottom()
+  M.scroll_to_bottom()
+end
+
+---Called when EventManager has finished emitting a batch of events
+function M.on_emit_events_finished()
+  M.scroll_to_bottom()
 end
 
 ---Auto-scroll to bottom if user was already at bottom
 ---Respects cursor position if user has scrolled up
-function M._scroll_to_bottom()
-  -- if we're loading a full session, don't scroll incrementally
-  if M._disable_auto_scroll then
-    return
-  end
-
+function M.scroll_to_bottom()
   local ok, line_count = pcall(vim.api.nvim_buf_line_count, state.windows.output_buf)
   if not ok then
     return
@@ -448,8 +442,6 @@ function M.on_message_updated(message, revert_index)
   end
 
   M._update_stats_from_message(msg)
-
-  M._scroll_to_bottom()
 end
 
 ---Event handler for message.part.updated events
@@ -526,8 +518,6 @@ function M.on_part_updated(properties, revert_index)
       M._rerender_part(text_part_id)
     end
   end
-
-  M._scroll_to_bottom()
 end
 
 ---Event handler for message.part.removed events
@@ -653,7 +643,6 @@ function M.on_permission_updated(permission)
   local part_id = M._find_part_by_call_id(permission.callID, permission.messageID)
   if part_id then
     M._rerender_part(part_id)
-    M._scroll_to_bottom()
   end
 end
 
@@ -672,7 +661,6 @@ function M.on_permission_replied(properties)
     local part_id = M._find_part_by_call_id(old_permission.callID, old_permission.messageID)
     if part_id then
       M._rerender_part(part_id)
-      M._scroll_to_bottom()
     end
   end
 end