diff --git a/.rubocop.yml b/.rubocop.yml
index 22d638c14..9de27ae45 100644
--- a/.rubocop.yml
+++ b/.rubocop.yml
@@ -258,6 +258,7 @@ Naming/PredicatePrefix:
 Metrics/ParameterLists:
   Exclude:
     - 'packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/base_collection.rb'
+    - 'packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/base_collection.rb'
     - 'packages/forest_admin_datasource_snowflake/lib/forest_admin_datasource_snowflake/datasource.rb'
     - 'packages/forest_admin_agent/lib/forest_admin_agent/routes/query_handler.rb'
     - 'packages/forest_admin_agent/lib/forest_admin_agent/services/smart_action_checker.rb'
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/client.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/client.rb
index 18121a53e..08678e409 100644
--- a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/client.rb
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/client.rb
@@ -1,5 +1,11 @@
 module ForestAdminDatasourcePylon
   class Client
+    MAX_SEARCH_LIMIT = 1000
+
+    # `next_cursor` is nil as soon as Pylon stops advertising a next page, so
+    # callers never have to know how the absence is spelled on the wire.
+    SearchPage = Struct.new(:records, :next_cursor, keyword_init: true)
+
     def initialize(configuration)
       @configuration = configuration
     end
@@ -10,8 +16,46 @@ def me
       must_succeed('me') { extract_data(connection.get('me').body) }
     end
 
+    # POST /issues/search accepts an empty body and then returns the most recent
+    # issues, ordered by `created_at` descending.
+    def search_issues(limit:, cursor: nil, filter: nil, search_text: nil)
+      body = { 'limit' => clamp_limit(limit) }
+      body['cursor']      = cursor unless blank?(cursor)
+      body['filter']      = filter unless filter.nil?
+      body['search_text'] = search_text unless blank?(search_text)
+
+      must_succeed('issues/search') { to_search_page(connection.post('issues/search', body).body) }
+    end
+
+    # Accepts either the UUID or the issue number. The id comes from
+    # operator-supplied filter values, so it is escaped before joining the path.
+    def fetch_issue(id)
+      path = "issues/#{Faraday::Utils.escape(id)}"
+      must_succeed(path) { extract_data(connection.get(path).body) }
+    end
+
     private
 
+    def clamp_limit(limit)
+      value = limit.to_i
+      return 1 if value < 1
+
+      [value, MAX_SEARCH_LIMIT].min
+    end
+
+    # Pylon only includes the `pagination` block when a next page exists, so an
+    # absent block, `has_next_page: false` and an empty cursor all mean "done".
+    def to_search_page(body)
+      pagination = body.is_a?(Hash) ? body['pagination'] : nil
+      cursor = pagination.is_a?(Hash) && pagination['has_next_page'] ? pagination['cursor'] : nil
+
+      SearchPage.new(records: Array(extract_data(body)), next_cursor: blank?(cursor) ? nil : cursor)
+    end
+
+    def blank?(value)
+      value.nil? || value.to_s.empty?
+    end
+
     # Pylon wraps payloads in { "data": ..., "pagination": ..., "request_id": ... }.
     def extract_data(body)
       return nil if body.nil? || body == ''
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/base_collection.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/base_collection.rb
new file mode 100644
index 000000000..24308e9df
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/base_collection.rb
@@ -0,0 +1,83 @@
+module ForestAdminDatasourcePylon
+  module Collections
+    class BaseCollection < ForestAdminDatasourceToolkit::Collection
+      ColumnSchema = ForestAdminDatasourceToolkit::Schema::ColumnSchema
+      Operators    = ForestAdminDatasourceToolkit::Components::Query::ConditionTree::Operators
+      Leaf         = ForestAdminDatasourceToolkit::Components::Query::ConditionTree::Nodes::ConditionTreeLeaf
+
+      STRING_OPS = [Operators::EQUAL, Operators::NOT_EQUAL, Operators::IN, Operators::NOT_IN,
+                    Operators::PRESENT, Operators::BLANK].freeze
+      NUMBER_OPS = (STRING_OPS + [Operators::GREATER_THAN, Operators::LESS_THAN]).freeze
+      DATE_OPS   = [Operators::EQUAL, Operators::BEFORE, Operators::AFTER,
+                    Operators::PRESENT, Operators::BLANK].freeze
+
+      attr_reader :custom_fields
+
+      # Template method: subclasses implement `define_schema` and
+      # `define_relations` as hooks; ordering between them, custom-field
+      # registration, and the search/count flags is owned here so collisions
+      # are always evaluated against the final native schema.
+      def initialize(datasource, name, custom_fields: [], searchable: false, countable: false, native_driver: nil)
+        super(datasource, name, native_driver)
+        define_schema
+        define_relations
+        @custom_fields = add_custom_fields(custom_fields)
+        enable_search if searchable
+        enable_count if countable
+      end
+
+      protected
+
+      # Pylon has no `id` filter operator on /issues/search, so collections
+      # short-circuit primary-key lookups to /resource/{id}. Ids are UUID
+      # strings — unlike Zendesk, nothing has to be coerced to an integer.
+      def extract_id_lookup(node)
+        return nil unless node.is_a?(Leaf) && node.field == 'id'
+
+        return unless [Operators::EQUAL, Operators::IN].include?(node.operator)
+
+        Array(node.value).map(&:to_s).reject(&:empty?)
+      end
+
+      def project(record, projection)
+        return record if projection.nil?
+
+        wanted = Array(projection).map(&:to_s).reject { |p| p.include?(':') }
+        return record if wanted.empty?
+
+        wanted.to_h { |k| [k, record[k]] }
+      end
+
+      def translate_page(page)
+        return [0, Client::MAX_SEARCH_LIMIT] if page.nil?
+
+        limit = page.limit.to_i.positive? ? page.limit.to_i : Client::MAX_SEARCH_LIMIT
+        [page.offset.to_i.clamp(0, nil), limit]
+      end
+
+      # Adds custom fields, skipping any whose column name collides with a
+      # field already declared on the collection. Returns the subset actually
+      # added so callers can keep their serializer in sync with the schema.
+      def add_custom_fields(custom_fields)
+        custom_fields.reject do |cf|
+          column_name = cf[:column_name]
+          if schema[:fields].key?(column_name)
+            ForestAdminDatasourcePylon.logger.warn(
+              "[forest_admin_datasource_pylon] Custom field '#{column_name}' on collection " \
+              "'#{name}' conflicts with an existing field; skipping."
+            )
+            true
+          else
+            add_field(column_name, cf[:schema])
+            false
+          end
+        end
+      end
+
+      private
+
+      def define_schema    = raise(NotImplementedError, "#{self.class} did not implement define_schema")
+      def define_relations = raise(NotImplementedError, "#{self.class} did not implement define_relations")
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue.rb
new file mode 100644
index 000000000..c7f139425
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue.rb
@@ -0,0 +1,67 @@
+module ForestAdminDatasourcePylon
+  module Collections
+    class Issue < BaseCollection
+      include SchemaDefinition
+      include Serializer
+
+      def initialize(datasource, custom_fields: [])
+        super(datasource, 'PylonIssue', custom_fields: custom_fields)
+      end
+
+      def list(_caller, filter, projection)
+        fetch_records(filter).map { |issue| project(serialize(issue), projection) }
+      end
+
+      private
+
+      def fetch_records(filter)
+        ids = extract_id_lookup(filter&.condition_tree)
+        return page_window(fetch_by_ids(ids), filter) if ids
+
+        warn_ignored_filter(filter)
+        offset, limit = translate_page(filter&.page)
+        walker.walk(offset: offset, limit: limit) do |batch, cursor|
+          datasource.client.search_issues(limit: batch, cursor: cursor)
+        end
+      end
+
+      # Sliced after the lookup, not before, so ids that resolved to nothing
+      # (404) do not eat into the requested window.
+      def page_window(records, filter)
+        offset, limit = translate_page(filter&.page)
+        records[offset, limit] || []
+      end
+
+      # A record the operator can no longer reach — deleted, or outside the
+      # token's scope — reads as "no record" rather than as a failed page.
+      def fetch_by_ids(ids)
+        ids.filter_map do |id|
+          datasource.client.fetch_issue(id)
+        rescue APIError => e
+          raise unless e.status == 404
+
+          nil
+        end
+      end
+
+      # Condition-tree translation and free-text search land in a later story.
+      # Until then a filter the collection cannot honour is dropped, which would
+      # otherwise silently return unfiltered rows.
+      def warn_ignored_filter(filter)
+        ignored = []
+        ignored << 'condition tree' unless filter&.condition_tree.nil?
+        ignored << 'search' unless filter&.search.nil? || filter.search.to_s.empty?
+        return if ignored.empty?
+
+        ForestAdminDatasourcePylon.logger.warn(
+          "[forest_admin_datasource_pylon] PylonIssue ignored the #{ignored.join(" and ")} of this query; " \
+          'filtering is not implemented yet, so the returned records are unfiltered.'
+        )
+      end
+
+      def walker
+        @walker ||= Pagination::CursorWalker.new
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/schema_definition.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/schema_definition.rb
new file mode 100644
index 000000000..d7d95edab
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/schema_definition.rb
@@ -0,0 +1,76 @@
+module ForestAdminDatasourcePylon
+  module Collections
+    class Issue < BaseCollection
+      # Every column is read-only and non-sortable in this story: writes land in
+      # a later story, and `/issues/search` exposes no sort parameter at all —
+      # results always come back ordered by `created_at` descending, so
+      # advertising a sortable column would let the UI ask for an order the API
+      # cannot honour.
+      module SchemaDefinition
+        ColumnSchema = BaseCollection::ColumnSchema
+        Operators    = BaseCollection::Operators
+        STRING_OPS   = BaseCollection::STRING_OPS
+        NUMBER_OPS   = BaseCollection::NUMBER_OPS
+        DATE_OPS     = BaseCollection::DATE_OPS
+
+        private
+
+        def define_schema
+          define_identity_fields
+          define_content_fields
+          define_party_fields
+          define_time_fields
+        end
+
+        # Relations are declared in a later story, once the Account / Contact /
+        # User / Team collections exist to point at.
+        def define_relations; end
+
+        def define_identity_fields
+          # Only equal/in: these are the two the primary-key short-circuit can
+          # actually serve through GET /issues/{id}.
+          add_field('id', ColumnSchema.new(column_type: 'String',
+                                           filter_operators: [Operators::EQUAL, Operators::IN],
+                                           is_primary_key: true, is_read_only: true))
+          add_field('number', column('Number', NUMBER_OPS))
+          add_field('link', column('String', []))
+        end
+
+        def define_content_fields
+          add_field('title', column('String', STRING_OPS))
+          add_field('body_html', column('String', []))
+          # Left as String rather than Enum: Pylon ships five built-in states
+          # but organisations define their own on top of them.
+          add_field('state', column('String', STRING_OPS))
+          add_field('type', column('String', STRING_OPS))
+          add_field('source', column('String', STRING_OPS))
+          add_field('tags', column('Json', []))
+          add_field('customer_portal_visible', column('Boolean', []))
+          add_field('author_unverified', column('Boolean', []))
+          add_field('number_of_touches', column('Number', NUMBER_OPS))
+        end
+
+        # Flattened from the nested `{id: …}` objects Pylon returns. They stay
+        # plain columns until the story that adds the related collections.
+        def define_party_fields
+          %w[account_id requester_id assignee_id team_id].each do |field|
+            add_field(field, column('String', STRING_OPS))
+          end
+        end
+
+        def define_time_fields
+          %w[first_response_time resolution_time latest_message_time created_at updated_at].each do |field|
+            add_field(field, column('Date', DATE_OPS))
+          end
+          %w[time_in_status_seconds business_hours_time_in_status_seconds].each do |field|
+            add_field(field, column('Json', []))
+          end
+        end
+
+        def column(type, operators)
+          ColumnSchema.new(column_type: type, filter_operators: operators, is_read_only: true)
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/serializer.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/serializer.rb
new file mode 100644
index 000000000..16a8bc511
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/serializer.rb
@@ -0,0 +1,45 @@
+module ForestAdminDatasourcePylon
+  module Collections
+    class Issue < BaseCollection
+      module Serializer
+        PARTY_FIELDS = { 'account_id' => 'account', 'requester_id' => 'requester',
+                         'assignee_id' => 'assignee', 'team_id' => 'team' }.freeze
+
+        NATIVE_FIELDS = %w[id number link title body_html state type source tags
+                           customer_portal_visible author_unverified number_of_touches
+                           first_response_time resolution_time latest_message_time
+                           created_at updated_at time_in_status_seconds
+                           business_hours_time_in_status_seconds].freeze
+
+        private
+
+        def serialize(issue)
+          attrs = issue.is_a?(Hash) ? issue : {}
+          record = NATIVE_FIELDS.to_h { |field| [field, attrs[field]] }
+          PARTY_FIELDS.each { |column, source| record[column] = nested_id(attrs[source]) }
+          add_custom_field_values(record, attrs['custom_fields'])
+          record
+        end
+
+        def nested_id(value)
+          value['id'] if value.is_a?(Hash)
+        end
+
+        def add_custom_field_values(record, values)
+          custom_fields.each do |cf|
+            entry = values.is_a?(Hash) ? values[cf[:column_name]] : nil
+            record[cf[:column_name]] = custom_field_value(entry)
+          end
+        end
+
+        # Pylon spells a custom field as `slug => {"slug": ..., "value": ...}`,
+        # with `"values": [...]` instead of `"value"` for multi-value fields.
+        def custom_field_value(entry)
+          return entry unless entry.is_a?(Hash)
+
+          entry.key?('value') ? entry['value'] : entry['values']
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/datasource.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/datasource.rb
new file mode 100644
index 000000000..c7574c05b
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/datasource.rb
@@ -0,0 +1,19 @@
+module ForestAdminDatasourcePylon
+  class Datasource < ForestAdminDatasourceToolkit::Datasource
+    attr_reader :client, :configuration
+
+    def initialize(api_key:, **options)
+      super()
+      @configuration = Configuration.new(api_key: api_key, **options)
+      @client = Client.new(@configuration)
+
+      register_collections
+    end
+
+    private
+
+    def register_collections
+      add_collection(Collections::Issue.new(self))
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/pagination/cursor_walker.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/pagination/cursor_walker.rb
new file mode 100644
index 000000000..ea086ace9
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/pagination/cursor_walker.rb
@@ -0,0 +1,73 @@
+module ForestAdminDatasourcePylon
+  module Pagination
+    # Forest asks for an offset/limit window; Pylon only knows how to hand out
+    # the next page of a cursor. Bridging the two means walking pages until the
+    # window is covered, then slicing. Deep offsets therefore cost one request
+    # per page, which is why the walk is capped: `/issues/search` allows 20
+    # requests per minute, so an unbounded walk would spend the whole budget of
+    # the agent on a single list view.
+    class CursorWalker
+      MAX_PAGES = 20
+      MAX_RECORDS = 5_000
+
+      def initialize(max_pages: MAX_PAGES, max_records: MAX_RECORDS)
+        @max_pages = max_pages
+        @max_records = max_records
+      end
+
+      # Yields `(limit, cursor)` and expects a Client::SearchPage back.
+      def walk(offset:, limit:)
+        return [] unless limit.to_i.positive?
+
+        needed = offset.to_i + limit.to_i
+        records = []
+        cursor = nil
+        pages = 0
+
+        loop do
+          page = yield(batch_size(needed, records.size), cursor)
+          records.concat(page.records)
+          pages += 1
+
+          break if stop?(page, cursor) || records.size >= needed
+
+          if capped?(pages, records.size)
+            log_truncation(offset: offset, limit: limit, pages: pages, collected: records.size)
+            break
+          end
+
+          cursor = page.next_cursor
+        end
+
+        records[offset.to_i, limit.to_i] || []
+      end
+
+      private
+
+      # An empty page or a cursor that does not move would loop forever; Pylon
+      # does neither today, but a walk driven by a remote value stops on its own
+      # terms rather than on the caps only.
+      def stop?(page, cursor)
+        page.next_cursor.nil? || page.records.empty? || page.next_cursor == cursor
+      end
+
+      def capped?(pages, collected)
+        pages >= @max_pages || collected >= @max_records
+      end
+
+      # Bounded by the window still missing and by the record budget left, so
+      # the walk never collects past @max_records.
+      def batch_size(needed, collected)
+        [needed - collected, @max_records - collected].min.clamp(1, Client::MAX_SEARCH_LIMIT)
+      end
+
+      def log_truncation(offset:, limit:, pages:, collected:)
+        ForestAdminDatasourcePylon.logger.warn(
+          "[forest_admin_datasource_pylon] Stopped paginating after #{pages} page(s) / #{collected} record(s) " \
+          "while fetching offset=#{offset} limit=#{limit}; results are truncated. " \
+          'Narrow the filter to reach records past this point.'
+        )
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/client_spec.rb b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/client_spec.rb
index 28a5ea9ee..08dbc0c48 100644
--- a/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/client_spec.rb
+++ b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/client_spec.rb
@@ -132,6 +132,124 @@ def json(payload, status = 200)
     end
   end
 
+  describe '#search_issues' do
+    it 'posts the limit and returns the records' do
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => [{ 'id' => 'i1' }]))
+
+      page = client.search_issues(limit: 2)
+
+      expect(page.records).to eq([{ 'id' => 'i1' }])
+      expect(WebMock).to have_requested(:post, "#{base}/issues/search").with(body: { 'limit' => 2 })
+    end
+
+    it 'omits cursor, filter and search_text when they are not provided' do
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => []))
+
+      client.search_issues(limit: 5, cursor: nil, filter: nil, search_text: '')
+
+      expect(WebMock).to have_requested(:post, "#{base}/issues/search").with(body: { 'limit' => 5 })
+    end
+
+    it 'forwards cursor, filter and search_text when provided' do
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => []))
+      filter = { 'field' => 'state', 'operator' => 'equals', 'values' => ['new'] }
+
+      client.search_issues(limit: 5, cursor: 'c1', filter: filter, search_text: 'boom')
+
+      expect(WebMock).to have_requested(:post, "#{base}/issues/search")
+        .with(body: { 'limit' => 5, 'cursor' => 'c1', 'filter' => filter, 'search_text' => 'boom' })
+    end
+
+    it 'clamps the limit to the API maximum' do
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => []))
+
+      client.search_issues(limit: 99_999)
+
+      expect(WebMock).to have_requested(:post, "#{base}/issues/search")
+        .with(body: { 'limit' => described_class::MAX_SEARCH_LIMIT })
+    end
+
+    it 'raises the limit to 1 when it is zero or negative' do
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => []))
+
+      client.search_issues(limit: 0)
+
+      expect(WebMock).to have_requested(:post, "#{base}/issues/search").with(body: { 'limit' => 1 })
+    end
+
+    it 'exposes the cursor when a next page is advertised' do
+      stub_request(:post, "#{base}/issues/search")
+        .to_return(json('data' => [], 'pagination' => { 'cursor' => 'c2', 'has_next_page' => true }))
+
+      expect(client.search_issues(limit: 1).next_cursor).to eq('c2')
+    end
+
+    # Pylon omits the block entirely on the last page, so absence is the
+    # common case rather than the edge case.
+    it 'reports no next cursor when the pagination block is absent' do
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => [{ 'id' => 'i1' }]))
+
+      expect(client.search_issues(limit: 1).next_cursor).to be_nil
+    end
+
+    it 'reports no next cursor when has_next_page is false' do
+      stub_request(:post, "#{base}/issues/search")
+        .to_return(json('data' => [], 'pagination' => { 'cursor' => 'c2', 'has_next_page' => false }))
+
+      expect(client.search_issues(limit: 1).next_cursor).to be_nil
+    end
+
+    it 'reports no next cursor when the advertised cursor is empty' do
+      stub_request(:post, "#{base}/issues/search")
+        .to_return(json('data' => [], 'pagination' => { 'cursor' => '', 'has_next_page' => true }))
+
+      expect(client.search_issues(limit: 1).next_cursor).to be_nil
+    end
+
+    it 'returns no records when the payload carries none' do
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => nil))
+
+      expect(client.search_issues(limit: 1).records).to eq([])
+    end
+
+    it 'wraps a failure in an APIError naming the endpoint' do
+      stub_request(:post, "#{base}/issues/search").to_return(json({ 'message' => 'bad filter' }, 400))
+
+      expect { client.search_issues(limit: 1) }
+        .to raise_error(ForestAdminDatasourcePylon::APIError, %r{issues/search: HTTP 400 bad filter})
+    end
+  end
+
+  describe '#fetch_issue' do
+    it 'unwraps the issue' do
+      stub_request(:get, "#{base}/issues/i1").to_return(json('data' => { 'id' => 'i1', 'number' => 1 }))
+
+      expect(client.fetch_issue('i1')).to eq('id' => 'i1', 'number' => 1)
+    end
+
+    it 'accepts an issue number as well as a uuid' do
+      stub_request(:get, "#{base}/issues/42").to_return(json('data' => { 'id' => 'i1', 'number' => 42 }))
+
+      expect(client.fetch_issue(42)).to include('number' => 42)
+    end
+
+    it 'escapes an id that would otherwise alter the request path' do
+      stub_request(:get, "#{base}/issues/..%2Fme").to_return(json('data' => nil))
+
+      client.fetch_issue('../me')
+
+      expect(WebMock).to have_requested(:get, "#{base}/issues/..%2Fme")
+    end
+
+    it 'wraps a missing issue in a 404 APIError' do
+      stub_request(:get, "#{base}/issues/nope").to_return(json({ 'message' => 'not found' }, 404))
+
+      expect { client.fetch_issue('nope') }.to raise_error(ForestAdminDatasourcePylon::APIError) { |error|
+        expect(error.status).to eq(404)
+      }
+    end
+  end
+
   describe 'rate limiting' do
     it 'retries a 429 and returns the eventual success' do
       stub_request(:get, "#{base}/me")
diff --git a/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/base_collection_spec.rb b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/base_collection_spec.rb
new file mode 100644
index 000000000..4daa26807
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/base_collection_spec.rb
@@ -0,0 +1,151 @@
+module ForestAdminDatasourcePylon
+  RSpec.describe Collections::BaseCollection do
+    def leaf(field, operator, value)
+      ForestAdminDatasourceToolkit::Components::Query::ConditionTree::Nodes::ConditionTreeLeaf
+        .new(field, operator, value)
+    end
+
+    def page(offset, limit)
+      ForestAdminDatasourceToolkit::Components::Query::Page.new(offset: offset, limit: limit)
+    end
+
+    let(:operators) { ForestAdminDatasourceToolkit::Components::Query::ConditionTree::Operators }
+
+    let(:datasource) do
+      instance_double(ForestAdminDatasourcePylon::Datasource,
+                      client: instance_double(ForestAdminDatasourcePylon::Client))
+    end
+
+    let(:subclass) do
+      Class.new(described_class) do
+        def define_schema; end
+        def define_relations; end
+
+        public :extract_id_lookup, :project, :translate_page, :add_custom_fields
+      end
+    end
+
+    let(:collection) { subclass.new(datasource, 'X') }
+
+    describe 'subclass contract' do
+      it 'raises NotImplementedError naming define_schema when the hook is missing' do
+        expect { Class.new(described_class).new(datasource, 'X') }
+          .to raise_error(NotImplementedError, /define_schema/)
+      end
+
+      it 'raises NotImplementedError naming define_relations when only define_schema is implemented' do
+        incomplete = Class.new(described_class) { def define_schema; end }
+
+        expect { incomplete.new(datasource, 'X') }.to raise_error(NotImplementedError, /define_relations/)
+      end
+    end
+
+    describe 'search/count flags' do
+      # Inverted from the Zendesk template: free-text search and Count land in
+      # a later story, so a collection has to opt in rather than opt out.
+      it 'leaves search and count disabled by default' do
+        expect(collection.is_searchable?).to be(false)
+        expect(collection.is_countable?).to be(false)
+      end
+
+      it 'honours searchable: true / countable: true from super' do
+        opted_in = subclass.new(datasource, 'X', searchable: true, countable: true)
+
+        expect(opted_in.is_searchable?).to be(true)
+        expect(opted_in.is_countable?).to be(true)
+      end
+    end
+
+    describe '#extract_id_lookup' do
+      it 'extracts a single id from an equality leaf' do
+        expect(collection.extract_id_lookup(leaf('id', operators::EQUAL, 'uuid-1'))).to eq(['uuid-1'])
+      end
+
+      it 'extracts every id from an in leaf' do
+        node = leaf('id', operators::IN, %w[uuid-1 uuid-2])
+
+        expect(collection.extract_id_lookup(node)).to eq(%w[uuid-1 uuid-2])
+      end
+
+      # Pylon ids are uuids: unlike Zendesk's integer ids, nothing is coerced.
+      it 'keeps the id as an opaque string' do
+        expect(collection.extract_id_lookup(leaf('id', operators::EQUAL, 42))).to eq(['42'])
+      end
+
+      it 'drops empty values' do
+        expect(collection.extract_id_lookup(leaf('id', operators::IN, ['uuid-1', '']))).to eq(['uuid-1'])
+      end
+
+      it 'ignores a leaf on another field' do
+        expect(collection.extract_id_lookup(leaf('state', operators::EQUAL, 'new'))).to be_nil
+      end
+
+      it 'ignores an operator the short-circuit cannot serve' do
+        expect(collection.extract_id_lookup(leaf('id', operators::NOT_EQUAL, 'uuid-1'))).to be_nil
+      end
+
+      it 'ignores a nil condition tree' do
+        expect(collection.extract_id_lookup(nil)).to be_nil
+      end
+    end
+
+    describe '#project' do
+      let(:record) { { 'id' => 'uuid-1', 'title' => 'Boom', 'state' => 'new' } }
+
+      it 'returns the record untouched when there is no projection' do
+        expect(collection.project(record, nil)).to eq(record)
+      end
+
+      it 'keeps only the projected fields' do
+        expect(collection.project(record, %w[id title])).to eq('id' => 'uuid-1', 'title' => 'Boom')
+      end
+
+      it 'yields nil for a projected field the record does not carry' do
+        expect(collection.project(record, %w[id missing])).to eq('id' => 'uuid-1', 'missing' => nil)
+      end
+
+      it 'ignores relation paths and returns the whole record when only those are asked for' do
+        expect(collection.project(record, ['account:name'])).to eq(record)
+      end
+    end
+
+    describe '#translate_page' do
+      it 'defaults to a single full-size page when Forest sends none' do
+        expect(collection.translate_page(nil)).to eq([0, Client::MAX_SEARCH_LIMIT])
+      end
+
+      it 'passes the offset and limit through' do
+        expect(collection.translate_page(page(10, 25))).to eq([10, 25])
+      end
+
+      it 'falls back to the maximum limit when the page carries none' do
+        expect(collection.translate_page(page(0, nil))).to eq([0, Client::MAX_SEARCH_LIMIT])
+      end
+
+      it 'clamps a negative offset to zero' do
+        expect(collection.translate_page(page(-5, 10))).to eq([0, 10])
+      end
+    end
+
+    describe '#add_custom_fields' do
+      let(:schema) { ForestAdminDatasourceToolkit::Schema::ColumnSchema.new(column_type: 'String') }
+
+      it 'adds a field and reports it as added' do
+        added = collection.add_custom_fields([{ column_name: 'severity', schema: schema }])
+
+        expect(added.map { |cf| cf[:column_name] }).to eq(['severity'])
+        expect(collection.fields).to have_key('severity')
+      end
+
+      it 'skips a field colliding with an existing one and warns' do
+        collection.add_field('severity', schema)
+        allow(ForestAdminDatasourcePylon.logger).to receive(:warn)
+
+        added = collection.add_custom_fields([{ column_name: 'severity', schema: schema }])
+
+        expect(added).to be_empty
+        expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).with(/conflicts with an existing field/)
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/issue_spec.rb b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/issue_spec.rb
new file mode 100644
index 000000000..36038e832
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/issue_spec.rb
@@ -0,0 +1,278 @@
+module ForestAdminDatasourcePylon
+  RSpec.describe Collections::Issue do
+    def filter(condition_tree: nil, search: nil, page: nil)
+      ForestAdminDatasourceToolkit::Components::Query::Filter.new(
+        condition_tree: condition_tree, search: search, page: page
+      )
+    end
+
+    def leaf(field, operator, value)
+      ForestAdminDatasourceToolkit::Components::Query::ConditionTree::Nodes::ConditionTreeLeaf
+        .new(field, operator, value)
+    end
+
+    def id_leaf(operator, value)
+      leaf('id', operator, value)
+    end
+
+    def page(offset, limit)
+      ForestAdminDatasourceToolkit::Components::Query::Page.new(offset: offset, limit: limit)
+    end
+
+    def json(payload, status = 200)
+      { status: status, body: payload.to_json, headers: { 'Content-Type' => 'application/json' } }
+    end
+
+    # Trimmed to the shape observed on the API: parties are nested objects
+    # carrying an id, and unset values come back as null rather than absent.
+    def issue_payload(id, overrides = {})
+      {
+        'id' => id, 'number' => 12, 'title' => 'Boom', 'link' => "https://app.usepylon.com/issues?issueNumber=#{id}",
+        'body_html' => '<p>boom</p>', 'state' => 'new', 'type' => 'ticket', 'source' => 'manual',
+        'account' => { 'id' => 'acc-1', 'external_ids' => nil }, 'requester' => { 'id' => 'req-1' },
+        'assignee' => nil, 'team' => nil, 'tags' => %w[urgent], 'custom_fields' => {},
+        'first_response_time' => nil, 'resolution_time' => nil, 'latest_message_time' => '2026-08-07T13:06:22Z',
+        'created_at' => '2026-08-07T13:06:22Z', 'updated_at' => '2026-08-07T13:06:22Z',
+        'customer_portal_visible' => false, 'number_of_touches' => 0, 'author_unverified' => false,
+        'time_in_status_seconds' => { 'open' => 701_211 },
+        'business_hours_time_in_status_seconds' => { 'open' => 172_799 }
+      }.merge(overrides)
+    end
+
+    let(:datasource) { ForestAdminDatasourcePylon::Datasource.new(api_key: 'k') }
+    let(:collection) { datasource.get_collection('PylonIssue') }
+    let(:base) { datasource.configuration.url }
+    let(:operators) { ForestAdminDatasourceToolkit::Components::Query::ConditionTree::Operators }
+
+    describe 'schema' do
+      it 'is named PylonIssue' do
+        expect(collection.name).to eq('PylonIssue')
+      end
+
+      it 'declares id as the primary key' do
+        expect(collection.fields['id'].is_primary_key).to be(true)
+      end
+
+      it 'only advertises the id operators the short-circuit can serve' do
+        expect(collection.fields['id'].filter_operators).to eq([operators::EQUAL, operators::IN])
+      end
+
+      it 'exposes the native columns observed on the API' do
+        expect(collection.fields.keys).to include(
+          'number', 'title', 'body_html', 'state', 'type', 'source', 'link', 'tags',
+          'account_id', 'requester_id', 'assignee_id', 'team_id',
+          'first_response_time', 'resolution_time', 'latest_message_time', 'created_at', 'updated_at',
+          'customer_portal_visible', 'author_unverified', 'number_of_touches',
+          'time_in_status_seconds', 'business_hours_time_in_status_seconds'
+        )
+      end
+
+      it 'types the response-time columns as dates, not durations' do
+        expect(collection.fields['first_response_time'].column_type).to eq('Date')
+        expect(collection.fields['resolution_time'].column_type).to eq('Date')
+      end
+
+      # /issues/search exposes no sort parameter, and writes land in a later story.
+      it 'declares every column read-only and non-sortable' do
+        expect(collection.fields.values.map(&:is_read_only).uniq).to eq([true])
+        expect(collection.fields.values.map(&:is_sortable).uniq).to eq([false])
+      end
+
+      it 'leaves search and count disabled' do
+        expect(collection.is_searchable?).to be(false)
+        expect(collection.is_countable?).to be(false)
+      end
+    end
+
+    describe '#list' do
+      it 'searches for the most recent issues and serializes them' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+
+        rows = collection.list(nil, filter, nil)
+
+        expect(rows.size).to eq(1)
+        expect(rows.first).to include('id' => 'i1', 'title' => 'Boom', 'state' => 'new',
+                                      'tags' => ['urgent'], 'number_of_touches' => 0)
+      end
+
+      it 'flattens the nested parties into foreign-key columns' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+
+        expect(collection.list(nil, filter, nil).first)
+          .to include('account_id' => 'acc-1', 'requester_id' => 'req-1',
+                      'assignee_id' => nil, 'team_id' => nil)
+      end
+
+      it 'keeps the nested objects out of the serialized record' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+
+        expect(collection.list(nil, filter, nil).first.keys)
+          .not_to include('account', 'requester', 'assignee', 'team')
+      end
+
+      it 'restricts the record to the projection' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+
+        expect(collection.list(nil, filter, %w[id title])).to eq([{ 'id' => 'i1', 'title' => 'Boom' }])
+      end
+
+      it 'forwards the requested page as the search limit' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+
+        collection.list(nil, filter(page: page(0, 1)), nil)
+
+        expect(WebMock).to have_requested(:post, "#{base}/issues/search").with(body: { 'limit' => 1 })
+      end
+
+      it 'returns an empty list when the search returns nothing' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => []))
+
+        expect(collection.list(nil, filter, nil)).to eq([])
+      end
+    end
+
+    describe '#list across cursor pages' do
+      it 'walks the cursor until the requested window is covered' do
+        stub_request(:post, "#{base}/issues/search")
+          .with(body: { 'limit' => 3 })
+          .to_return(json('data' => [issue_payload('i1'), issue_payload('i2')],
+                          'pagination' => { 'cursor' => 'c1', 'has_next_page' => true }))
+        stub_request(:post, "#{base}/issues/search")
+          .with(body: { 'limit' => 1, 'cursor' => 'c1' })
+          .to_return(json('data' => [issue_payload('i3')]))
+
+        rows = collection.list(nil, filter(page: page(2, 1)), %w[id])
+
+        expect(rows).to eq([{ 'id' => 'i3' }])
+      end
+
+      it 'stops on the last page even when the window is not filled' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+
+        rows = collection.list(nil, filter(page: page(0, 50)), %w[id])
+
+        expect(rows).to eq([{ 'id' => 'i1' }])
+        expect(WebMock).to have_requested(:post, "#{base}/issues/search").once
+      end
+    end
+
+    describe '#list on a primary-key lookup' do
+      it 'reads the issue directly instead of searching' do
+        stub_request(:get, "#{base}/issues/i1").to_return(json('data' => issue_payload('i1')))
+
+        rows = collection.list(nil, filter(condition_tree: id_leaf(operators::EQUAL, 'i1')), %w[id title])
+
+        expect(rows).to eq([{ 'id' => 'i1', 'title' => 'Boom' }])
+        expect(WebMock).not_to have_requested(:post, "#{base}/issues/search")
+      end
+
+      it 'reads every id of an in filter, preserving their order' do
+        stub_request(:get, "#{base}/issues/i1").to_return(json('data' => issue_payload('i1')))
+        stub_request(:get, "#{base}/issues/i2").to_return(json('data' => issue_payload('i2')))
+
+        rows = collection.list(nil, filter(condition_tree: id_leaf(operators::IN, %w[i1 i2])), %w[id])
+
+        expect(rows).to eq([{ 'id' => 'i1' }, { 'id' => 'i2' }])
+      end
+
+      it 'applies the requested page to an in lookup' do
+        %w[i1 i2 i3].each do |id|
+          stub_request(:get, "#{base}/issues/#{id}").to_return(json('data' => issue_payload(id)))
+        end
+        query = filter(condition_tree: id_leaf(operators::IN, %w[i1 i2 i3]), page: page(1, 1))
+
+        expect(collection.list(nil, query, %w[id])).to eq([{ 'id' => 'i2' }])
+      end
+
+      it 'slices the page over the records that still exist' do
+        stub_request(:get, "#{base}/issues/gone").to_return(json({ 'message' => 'not found' }, 404))
+        %w[i2 i3].each do |id|
+          stub_request(:get, "#{base}/issues/#{id}").to_return(json('data' => issue_payload(id)))
+        end
+        query = filter(condition_tree: id_leaf(operators::IN, %w[gone i2 i3]), page: page(0, 2))
+
+        expect(collection.list(nil, query, %w[id])).to eq([{ 'id' => 'i2' }, { 'id' => 'i3' }])
+      end
+
+      it 'skips an issue that no longer exists' do
+        stub_request(:get, "#{base}/issues/i1").to_return(json('data' => issue_payload('i1')))
+        stub_request(:get, "#{base}/issues/gone").to_return(json({ 'message' => 'not found' }, 404))
+
+        rows = collection.list(nil, filter(condition_tree: id_leaf(operators::IN, %w[i1 gone])), %w[id])
+
+        expect(rows).to eq([{ 'id' => 'i1' }])
+      end
+
+      it 'propagates a failure that is not a missing record' do
+        stub_request(:get, "#{base}/issues/i1").to_return(json({ 'message' => 'boom' }, 500))
+
+        expect { collection.list(nil, filter(condition_tree: id_leaf(operators::EQUAL, 'i1')), %w[id]) }
+          .to raise_error(APIError)
+      end
+    end
+
+    describe 'custom fields' do
+      let(:column) { ForestAdminDatasourceToolkit::Schema::ColumnSchema.new(column_type: 'String') }
+      let(:collection) do
+        described_class.new(datasource, custom_fields: [{ column_name: 'severity', schema: column },
+                                                        { column_name: 'zones', schema: column }])
+      end
+
+      it 'serializes single- and multi-value custom fields' do
+        fields = { 'severity' => { 'slug' => 'severity', 'value' => 'high' },
+                   'zones' => { 'slug' => 'zones', 'values' => %w[eu us] } }
+        stub_request(:post, "#{base}/issues/search")
+          .to_return(json('data' => [issue_payload('i1', 'custom_fields' => fields)]))
+
+        expect(collection.list(nil, filter, nil).first).to include('severity' => 'high', 'zones' => %w[eu us])
+      end
+
+      it 'yields nil for a custom field the issue does not carry' do
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+
+        expect(collection.list(nil, filter, nil).first).to include('severity' => nil, 'zones' => nil)
+      end
+    end
+
+    describe 'filters it cannot honour yet' do
+      before do
+        allow(ForestAdminDatasourcePylon.logger).to receive(:warn)
+        stub_request(:post, "#{base}/issues/search").to_return(json('data' => [issue_payload('i1')]))
+      end
+
+      it 'warns and returns unfiltered records for a non primary-key condition' do
+        tree = leaf('state', operators::EQUAL, 'closed')
+
+        expect(collection.list(nil, filter(condition_tree: tree), %w[id])).to eq([{ 'id' => 'i1' }])
+        expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).with(/ignored the condition tree/)
+      end
+
+      it 'warns when a free-text search is supplied' do
+        collection.list(nil, filter(search: 'boom'), %w[id])
+
+        expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).with(/ignored the search/)
+      end
+
+      it 'names both when a condition tree and a search are supplied' do
+        tree = leaf('state', operators::EQUAL, 'closed')
+
+        collection.list(nil, filter(condition_tree: tree, search: 'boom'), %w[id])
+
+        expect(ForestAdminDatasourcePylon.logger)
+          .to have_received(:warn).with(/ignored the condition tree and search/)
+      end
+
+      it 'stays quiet when nothing was dropped' do
+        collection.list(nil, filter, %w[id])
+
+        expect(ForestAdminDatasourcePylon.logger).not_to have_received(:warn)
+      end
+
+      it 'stays quiet on an empty search string' do
+        collection.list(nil, filter(search: ''), %w[id])
+
+        expect(ForestAdminDatasourcePylon.logger).not_to have_received(:warn)
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/datasource_spec.rb b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/datasource_spec.rb
new file mode 100644
index 000000000..568251689
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/datasource_spec.rb
@@ -0,0 +1,36 @@
+RSpec.describe ForestAdminDatasourcePylon::Datasource do
+  let(:datasource) { described_class.new(api_key: 'k') }
+
+  it 'builds a configuration from the api key' do
+    expect(datasource.configuration.api_key).to eq('k')
+    expect(datasource.configuration.url).to eq(ForestAdminDatasourcePylon::Configuration::DEFAULT_BASE_URL)
+  end
+
+  it 'forwards the remaining options to the configuration' do
+    custom = described_class.new(api_key: 'k', base_url: 'https://pylon.test/', timeout: 3)
+
+    expect(custom.configuration.url).to eq('https://pylon.test')
+    expect(custom.configuration.timeout).to eq(3)
+  end
+
+  it 'exposes a client built on that configuration' do
+    expect(datasource.client).to be_a(ForestAdminDatasourcePylon::Client)
+  end
+
+  it 'registers the issue collection' do
+    expect(datasource.collections.keys).to eq(['PylonIssue'])
+    expect(datasource.get_collection('PylonIssue')).to be_a(ForestAdminDatasourcePylon::Collections::Issue)
+  end
+
+  it 'refuses to build without an api key' do
+    expect { described_class.new(api_key: nil) }.to raise_error(ForestAdminDatasourcePylon::ConfigurationError)
+  end
+
+  # Nothing is introspected at boot yet: registering collections must not hit
+  # the API, so an agent boots even when Pylon is unreachable.
+  it 'does not call the API while registering collections' do
+    datasource
+
+    expect(WebMock).not_to have_requested(:any, /usepylon/)
+  end
+end
diff --git a/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/pagination/cursor_walker_spec.rb b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/pagination/cursor_walker_spec.rb
new file mode 100644
index 000000000..116d4a897
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/pagination/cursor_walker_spec.rb
@@ -0,0 +1,113 @@
+RSpec.describe ForestAdminDatasourcePylon::Pagination::CursorWalker do
+  let(:calls) { [] }
+
+  def search_page(ids, next_cursor)
+    ForestAdminDatasourcePylon::Client::SearchPage.new(
+      records: ids.map { |id| { 'id' => id } }, next_cursor: next_cursor
+    )
+  end
+
+  # Serves `pages` in order and records the (limit, cursor) each call was made
+  # with, so a spec can assert on how the walk was driven.
+  def source(pages)
+    proc do |limit, cursor|
+      calls << { limit: limit, cursor: cursor }
+      pages.fetch(calls.size - 1, search_page([], nil))
+    end
+  end
+
+  def walk(pages, offset:, limit:, walker: described_class.new)
+    walker.walk(offset: offset, limit: limit, &source(pages))
+  end
+
+  it 'returns the records of a single page when it covers the window' do
+    records = walk([search_page(%w[a b c], nil)], offset: 0, limit: 3)
+
+    expect(records).to eq([{ 'id' => 'a' }, { 'id' => 'b' }, { 'id' => 'c' }])
+    expect(calls).to eq([{ limit: 3, cursor: nil }])
+  end
+
+  it 'walks pages until the window is covered, then slices off the offset' do
+    pages = [search_page(%w[a b], 'c1'), search_page(%w[c d], 'c2'), search_page(%w[e f], nil)]
+
+    expect(walk(pages, offset: 4, limit: 2)).to eq([{ 'id' => 'e' }, { 'id' => 'f' }])
+    expect(calls.map { |call| call[:cursor] }).to eq([nil, 'c1', 'c2'])
+  end
+
+  it 'asks only for what is still missing on each page' do
+    pages = [search_page(%w[a b], 'c1'), search_page(%w[c], 'c2'), search_page(%w[d], nil)]
+
+    walk(pages, offset: 0, limit: 4)
+
+    expect(calls.map { |call| call[:limit] }).to eq([4, 2, 1])
+  end
+
+  it 'never requests more than the API maximum in one call' do
+    walk([search_page(%w[a], nil)], offset: 0, limit: 50_000)
+
+    expect(calls.first[:limit]).to eq(ForestAdminDatasourcePylon::Client::MAX_SEARCH_LIMIT)
+  end
+
+  it 'stops at the last page and returns fewer records than asked' do
+    expect(walk([search_page(%w[a b], nil)], offset: 0, limit: 10).size).to eq(2)
+    expect(calls.size).to eq(1)
+  end
+
+  it 'returns an empty window when the offset is past the end' do
+    expect(walk([search_page(%w[a b], nil)], offset: 50, limit: 10)).to eq([])
+  end
+
+  it 'fetches nothing when the limit is not positive' do
+    expect(walk([search_page(%w[a], nil)], offset: 0, limit: 0)).to eq([])
+    expect(calls).to be_empty
+  end
+
+  describe 'truncation' do
+    before { allow(ForestAdminDatasourcePylon.logger).to receive(:warn) }
+
+    it 'stops at the page cap and logs a warning' do
+      pages = Array.new(5) { |i| search_page(["r#{i}"], "c#{i}") }
+
+      expect(walk(pages, offset: 0, limit: 100, walker: described_class.new(max_pages: 3)).size).to eq(3)
+      expect(calls.size).to eq(3)
+      expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).with(/Stopped paginating after 3 page/)
+    end
+
+    it 'stops at the record cap and logs a warning' do
+      pages = Array.new(5) { |i| search_page(%W[a#{i} b#{i}], "c#{i}") }
+
+      expect(walk(pages, offset: 0, limit: 100, walker: described_class.new(max_records: 4)).size).to eq(4)
+      expect(ForestAdminDatasourcePylon.logger).to have_received(:warn).with(/4 record\(s\)/)
+    end
+
+    it 'never asks for more than the remaining record budget' do
+      pages = [search_page(%w[a b], 'c1'), search_page(%w[c], 'c2')]
+
+      walk(pages, offset: 0, limit: 100, walker: described_class.new(max_records: 3))
+
+      expect(calls.map { |call| call[:limit] }).to eq([3, 1])
+    end
+
+    it 'does not warn when the walk ends naturally' do
+      walk([search_page(%w[a b], nil)], offset: 0, limit: 100)
+
+      expect(ForestAdminDatasourcePylon.logger).not_to have_received(:warn)
+    end
+  end
+
+  describe 'defensive stops' do
+    it 'stops when a page comes back empty despite advertising a next page' do
+      pages = [search_page([], 'c1'), search_page(%w[a], 'c2')]
+
+      expect(walk(pages, offset: 0, limit: 10)).to eq([])
+      expect(calls.size).to eq(1)
+    end
+
+    it 'stops when the cursor does not advance' do
+      pages = [search_page(%w[a], 'same'), search_page(%w[b], 'same'), search_page(%w[c], 'same')]
+
+      expect(walk(pages, offset: 0, limit: 10).size).to eq(2)
+      expect(calls.size).to eq(2)
+    end
+  end
+end