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 80026c695..4e629245f 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
@@ -4,6 +4,12 @@ module ForestAdminDatasourcePylon
   class Client # rubocop:disable Metrics/ClassLength
     MAX_SEARCH_LIMIT = 1000
 
+    # Bounds `collect_pages`, which asks for a whole dataset rather than a
+    # window: the endpoints it reads answer in one response, so reaching this
+    # many pages means the API started paginating on its own and the walk is
+    # spending more of the per-minute budget than the answer is worth.
+    MAX_COLLECTED_PAGES = 10
+
     # `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)
@@ -29,6 +35,23 @@ def fetch_issue(id)
       fetch_resource('issues', id)
     end
 
+    # The whole conversation of an issue, oldest message first, or nil when the
+    # thread could not be read.
+    #
+    # `limit` is left out on purpose: Pylon then answers with every message in a
+    # single response. Asking for a page would hand back the OLDEST messages and
+    # cut the most recent ones off, which is the half of a conversation nobody
+    # opens a ticket to read.
+    #
+    # The cursor is still followed, defensively: Pylon paginates this endpoint
+    # when asked to, so a future default page size stays handled rather than
+    # silently truncating the thread.
+    def fetch_issue_messages(issue_id)
+      path = "issues/#{Faraday::Utils.escape(issue_id)}/messages"
+
+      best_effort("fetch_issue_messages(#{issue_id})", default: nil) { must_succeed(path) { collect_pages(path) } }
+    end
+
     def search_accounts(limit:, cursor: nil, filter: nil, search_text: nil)
       search_resource('accounts/search', limit: limit, cursor: cursor, filter: filter, search_text: search_text)
     end
@@ -99,6 +122,41 @@ def fetch_all(path, params = {})
       must_succeed(path) { Array(extract_data(connection.get(path, params).body)) }
     end
 
+    # Every record of a cursor-paginated GET, no window asked for and no limit
+    # sent. `CursorWalker` answers the other question — the offset/limit window a
+    # list view asks for — and is not what this needs.
+    #
+    # An empty page and a cursor that does not move both stop the loop: neither
+    # happens today, but a walk driven by a remote value stops on its own terms.
+    def collect_pages(path)
+      records = []
+      cursor = nil
+      pages = 0
+
+      loop do
+        page = to_search_page(connection.get(path, cursor.nil? ? {} : { 'cursor' => cursor }).body)
+        records.concat(page.records)
+        pages += 1
+        break if page.next_cursor.nil? || page.next_cursor == cursor || page.records.empty?
+
+        if pages >= MAX_COLLECTED_PAGES
+          log_pagination_cap(path, pages, records.size)
+          break
+        end
+
+        cursor = page.next_cursor
+      end
+
+      records
+    end
+
+    def log_pagination_cap(path, pages, collected)
+      ForestAdminDatasourcePylon.logger.warn(
+        "[forest_admin_datasource_pylon] Stopped paginating #{path} after #{pages} page(s) / " \
+        "#{collected} record(s); the rest is left out."
+      )
+    end
+
     # The id comes from operator-supplied filter values, so it is escaped before
     # being joined to the path.
     def fetch_resource(resource, id)
@@ -145,6 +203,18 @@ def must_succeed(operation)
       raise APIError, "Pylon API call failed: #{operation}: #{e.class}: #{e.message}"
     end
 
+    # For the calls whose result enriches a page rather than being the page: the
+    # failure is reported and the default returned, so a degraded thread or a
+    # missing enrichment costs the operator a column, not the record they opened.
+    def best_effort(operation, default:)
+      yield
+    rescue StandardError => e
+      ForestAdminDatasourcePylon.logger.warn(
+        "[forest_admin_datasource_pylon] #{operation} failed; degrading: #{e.class}: #{e.message}"
+      )
+      default
+    end
+
     # Builds an APIError preserving the HTTP status and Pylon's own error body so
     # smart actions can show the operator the real reason instead of "failed".
     def api_error(operation, error)
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
index e6261cad8..fdf48b40f 100644
--- 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
@@ -5,6 +5,7 @@ class Issue < BaseCollection
       include RecordSerialization
       include Serializer
       include RelationEmbedder
+      include MessagesEmbedder
 
       # `/issues/search` exposes no sort parameter, so the allow-list is empty and
       # every requested order is reported instead of being silently swallowed.
@@ -19,6 +20,32 @@ class Issue < BaseCollection
       # throttling that would let this cap grow.
       MAX_ID_LOOKUPS = 20
 
+      # The shape of one message inside the `messages` column. Field names follow
+      # the columns of this collection rather than the payload: Pylon spells them
+      # `message_html` and `timestamp`, which would put two conventions in the
+      # same schema for the operator to reconcile.
+      MESSAGE_THREAD_SCHEMA = {
+        'id' => 'String',
+        'body_html' => 'String',
+        'is_private' => 'Boolean',
+        'source' => 'String',
+        'thread_id' => 'String',
+        'file_urls' => 'Json',
+        'created_at' => 'Date',
+        'author_name' => 'String',
+        'author_email' => 'String',
+        'author_avatar_url' => 'String',
+        'author_contact_id' => 'String',
+        'author_user_id' => 'String'
+      }.freeze
+
+      # One thread is one `GET /issues/{id}/messages`, an endpoint allowing 20
+      # requests per minute: a page asking for more threads than this reads the
+      # first ones and reports the rest, for the same reason MAX_ID_LOOKUPS
+      # bounds the primary-key fan-out. Story 9 (EXT-13) owns the throttling
+      # that would let this cap grow.
+      MAX_MESSAGE_EMBEDS = 10
+
       def initialize(datasource, custom_fields: [])
         super(datasource, 'PylonIssue', custom_fields: custom_fields, searchable: true)
       end
@@ -27,6 +54,7 @@ def list(caller, filter, projection)
         records = fetch_records(caller, filter)
         rows = records.map { |record| project(record, projection) }
         embed_relations(records, rows, projection)
+        embed_messages(records, rows) if want_messages?(projection)
         rows
       end
 
diff --git a/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/messages_embedder.rb b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/messages_embedder.rb
new file mode 100644
index 000000000..11762add8
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/lib/forest_admin_datasource_pylon/collections/issue/messages_embedder.rb
@@ -0,0 +1,89 @@
+module ForestAdminDatasourcePylon
+  module Collections
+    class Issue < BaseCollection
+      # The conversation of an issue, embedded as a structured array column the
+      # way the Zendesk datasource embeds a ticket's comments.
+      #
+      # Pylon has no way to read the threads of several issues at once, so the
+      # thread costs one request per row — against an endpoint allowing 20 per
+      # minute. The fan-out is therefore bounded like the primary-key lookups of
+      # this collection: truncated with a warning rather than turned into a rate
+      # limit error halfway through the page. A relation the projection does not
+      # ask for costs no request at all.
+      module MessagesEmbedder
+        include RecordSerialization
+
+        private
+
+        # The thread is read only when the projection names it. A nil projection
+        # — what a count or an export goes through — asks for the record as
+        # Pylon returns it, and embeds nothing, exactly like RelationEmbedder:
+        # spending one request per row on a path that never asked for the
+        # conversation is the very fan-out MAX_MESSAGE_EMBEDS exists to bound.
+        def want_messages?(projection)
+          Array(projection).map(&:to_s).any? { |p| p == 'messages' || p.start_with?('messages:') }
+        end
+
+        # A row past the cap, and a row whose thread failed to be read, are left
+        # at nil: "unknown", never the empty list, which would read as "this
+        # issue has no message" — the kind of answer that looks complete without
+        # being it.
+        def embed_messages(records, rows)
+          embedded = rows.first(MAX_MESSAGE_EMBEDS)
+          warn_truncated_threads(rows.size) if rows.size > embedded.size
+
+          embedded.each_with_index do |row, index|
+            messages = datasource.client.fetch_issue_messages(records[index]['id'])
+            row['messages'] = messages&.map { |message| serialize_message(message) }
+          end
+        end
+
+        def serialize_message(message)
+          attrs = message.is_a?(Hash) ? message : {}
+
+          {
+            'id' => attrs['id'],
+            'body_html' => attrs['message_html'],
+            'is_private' => attrs['is_private'],
+            'source' => attrs['source'],
+            'thread_id' => attrs['thread_id'],
+            'file_urls' => attrs['file_urls'],
+            'created_at' => attrs['timestamp']
+          }.merge(flatten_author(attrs['author']))
+        end
+
+        # Pylon nests the author's contact and user sides side by side, both
+        # optional and with nothing telling them apart: a message written by an
+        # agent carries `user`, one written by a customer carries `contact`. Both
+        # ids are kept, so a message stays traceable to the PylonContact or
+        # PylonUser record it came from, and the email is taken from whichever
+        # side is there.
+        def flatten_author(author)
+          attrs   = author.is_a?(Hash) ? author : {}
+          contact = attrs['contact']
+          user    = attrs['user']
+
+          {
+            'author_name' => attrs['name'],
+            'author_avatar_url' => attrs['avatar_url'],
+            'author_email' => nested_email(contact) || nested_email(user),
+            'author_contact_id' => nested_id(contact),
+            'author_user_id' => nested_id(user)
+          }
+        end
+
+        def nested_email(value)
+          value['email'] if value.is_a?(Hash)
+        end
+
+        def warn_truncated_threads(asked)
+          ForestAdminDatasourcePylon.logger.warn(
+            "[forest_admin_datasource_pylon] Asked for the message thread of #{asked} issues, reading the first " \
+            "#{MAX_MESSAGE_EMBEDS}: one request per issue would exhaust the rate limit of the agent. " \
+            'Narrow the selection, or take the thread out of the projection, to reach the records past this point.'
+          )
+        end
+      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
index 924d82da5..838d04045 100644
--- 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
@@ -68,6 +68,23 @@ def define_content_fields
           add_column('customer_portal_visible', 'Boolean')
           add_column('author_unverified', 'Boolean')
           add_column('number_of_touches', 'Number')
+          define_thread_field
+        end
+
+        # The conversation, embedded at read time by MessagesEmbedder. Declared
+        # by hand rather than through `add_column`: its type is the shape of one
+        # message, not a primitive.
+        #
+        # Neither filterable nor sortable — `POST /issues/search` covers no
+        # message field, and the thread is not even part of the payload the
+        # search endpoint returns — and not groupable either: `ColumnSchema`
+        # defaults that flag to true, where the `add_column` of the base passes
+        # false for every Pylon column, a thread being both an array and a value
+        # the pages of a cursor walk do not carry.
+        def define_thread_field
+          add_field('messages', ColumnSchema.new(column_type: [Issue::MESSAGE_THREAD_SCHEMA],
+                                                 filter_operators: [], is_groupable: false,
+                                                 is_read_only: true))
         end
 
         # Flattened from the nested `{id: …}` objects Pylon returns, and kept as
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 f0766e309..d52037f84 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
@@ -250,6 +250,95 @@ def json(payload, status = 200)
     end
   end
 
+  describe '#fetch_issue_messages' do
+    let(:logger) { instance_double(Logger, warn: nil) }
+
+    def page(records, cursor: nil)
+      body = { 'data' => records }
+      body['pagination'] = { 'cursor' => cursor, 'has_next_page' => true } if cursor
+      json(body)
+    end
+
+    it 'returns the whole thread of an issue' do
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(page([{ 'id' => 'm1' }, { 'id' => 'm2' }]))
+
+      expect(client.fetch_issue_messages('i1')).to eq([{ 'id' => 'm1' }, { 'id' => 'm2' }])
+    end
+
+    # Omitting `limit` is what makes Pylon answer with every message at once;
+    # asking for a page would hand back the oldest ones and cut off the rest.
+    it 'sends no limit, so Pylon answers with every message in one request' do
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(page([]))
+
+      client.fetch_issue_messages('i1')
+
+      expect(WebMock).to have_requested(:get, "#{base}/issues/i1/messages").with(query: {}).once
+    end
+
+    it 'escapes an id that would otherwise alter the request path' do
+      stub_request(:get, "#{base}/issues/..%2Fme/messages").to_return(page([]))
+
+      client.fetch_issue_messages('../me')
+
+      expect(WebMock).to have_requested(:get, "#{base}/issues/..%2Fme/messages")
+    end
+
+    it 'follows the cursor when Pylon paginates the thread anyway' do
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: {})
+                                                      .to_return(page([{ 'id' => 'm1' }], cursor: 'c1'))
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: { 'cursor' => 'c1' })
+                                                      .to_return(page([{ 'id' => 'm2' }]))
+
+      expect(client.fetch_issue_messages('i1')).to eq([{ 'id' => 'm1' }, { 'id' => 'm2' }])
+    end
+
+    it 'stops on a cursor that does not move' do
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: { 'cursor' => 'c1' })
+                                                      .to_return(page([{ 'id' => 'm2' }], cursor: 'c1'))
+      stub_request(:get, "#{base}/issues/i1/messages").with(query: {})
+                                                      .to_return(page([{ 'id' => 'm1' }], cursor: 'c1'))
+
+      expect(client.fetch_issue_messages('i1').size).to eq(2)
+    end
+
+    it 'stops on an empty page' do
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(page([], cursor: 'c1'))
+
+      expect(client.fetch_issue_messages('i1')).to eq([])
+      expect(WebMock).to have_requested(:get, "#{base}/issues/i1/messages").once
+    end
+
+    it 'caps a thread Pylon never stops paginating, and says so' do
+      allow(ForestAdminDatasourcePylon).to receive(:logger).and_return(logger)
+      served = 0
+      stub_request(:get, %r{/issues/i1/messages}).to_return do
+        served += 1
+        page([{ 'id' => "m#{served}" }], cursor: "c#{served}")
+      end
+
+      client.fetch_issue_messages('i1')
+
+      expect(served).to eq(described_class::MAX_COLLECTED_PAGES)
+      expect(logger).to have_received(:warn).with(/Stopped paginating/)
+    end
+
+    # The thread enriches a page rather than being it: a failure costs the
+    # operator the column, not the records they opened.
+    it 'degrades to nil and reports the failure when the thread cannot be read' do
+      allow(ForestAdminDatasourcePylon).to receive(:logger).and_return(logger)
+      stub_request(:get, "#{base}/issues/i1/messages").to_return(json({ 'message' => 'boom' }, 500))
+
+      expect(client.fetch_issue_messages('i1')).to be_nil
+      expect(logger).to have_received(:warn).with(/fetch_issue_messages\(i1\) failed; degrading.*HTTP 500 boom/)
+    end
+
+    it 'degrades on a missing issue rather than raising a 404' do
+      stub_request(:get, "#{base}/issues/nope/messages").to_return(json({ 'message' => 'not found' }, 404))
+
+      expect(client.fetch_issue_messages('nope')).to be_nil
+    end
+  end
+
   describe '#search_accounts' do
     it 'posts the full search envelope and returns the records' do
       stub_request(:post, "#{base}/accounts/search").to_return(json('data' => [{ 'id' => 'a1' }]))
diff --git a/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/issue/messages_embedder_spec.rb b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/issue/messages_embedder_spec.rb
new file mode 100644
index 000000000..51f03f241
--- /dev/null
+++ b/packages/forest_admin_datasource_pylon/spec/forest_admin_datasource_pylon/collections/issue/messages_embedder_spec.rb
@@ -0,0 +1,240 @@
+module ForestAdminDatasourcePylon
+  # Observed through PylonIssue, the only collection carrying a conversation.
+  RSpec.describe Collections::Issue::MessagesEmbedder 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 json(payload, status = 200)
+      { status: status, body: payload.to_json, headers: { 'Content-Type' => 'application/json' } }
+    end
+
+    def issue_payload(id, overrides = {})
+      { 'id' => id, 'title' => 'Boom' }.merge(overrides)
+    end
+
+    # Trimmed to the shape observed on the API: the author carries a contact or
+    # a user side, and unset values come back as null rather than absent.
+    def message_payload(id, overrides = {})
+      {
+        'id' => id, 'message_html' => '<p>hello</p>', 'is_private' => false, 'source' => 'email',
+        'thread_id' => nil, 'file_urls' => [], 'timestamp' => '2026-08-07T13:06:22Z',
+        'author' => { 'name' => 'Ada', 'avatar_url' => 'https://cdn/ada.png',
+                      'contact' => { 'id' => 'con-1', 'email' => 'ada@acme.com' }, 'user' => nil }
+      }.merge(overrides)
+    end
+
+    def stub_issues(*payloads)
+      stub_request(:post, "#{base}/issues/search").to_return(json('data' => payloads))
+    end
+
+    def stub_messages(issue_id, *payloads)
+      stub_request(:get, "#{base}/issues/#{issue_id}/messages").to_return(json('data' => payloads))
+    end
+
+    let(:datasource) { ForestAdminDatasourcePylon::Datasource.new(api_key: 'k') }
+    let(:issues) { datasource.get_collection('PylonIssue') }
+    let(:base) { datasource.configuration.url }
+    let(:logger) { instance_double(Logger, warn: nil) }
+
+    describe 'the schema of the column' do
+      it 'declares messages as an array of message shapes' do
+        expect(issues.fields['messages'].column_type).to eq([Collections::Issue::MESSAGE_THREAD_SCHEMA])
+      end
+
+      it 'names the fields after the columns of the collection, not after the payload' do
+        expect(Collections::Issue::MESSAGE_THREAD_SCHEMA.keys).to include('body_html', 'created_at')
+        expect(Collections::Issue::MESSAGE_THREAD_SCHEMA.keys).not_to include('message_html', 'timestamp')
+      end
+
+      it 'advertises no filter and no sort on a thread the search endpoint does not cover' do
+        expect(issues.fields['messages'].filter_operators).to eq([])
+        expect(issues.fields['messages'].is_sortable).to be(false)
+      end
+
+      it 'is read-only, like every other column of this story' do
+        expect(issues.fields['messages'].is_read_only).to be(true)
+      end
+    end
+
+    describe 'what the projection asks for' do
+      before { stub_issues(issue_payload('i1')) }
+
+      it 'reads the thread when the projection names it' do
+        stub_messages('i1', message_payload('msg-1'))
+
+        rows = issues.list(nil, filter, %w[id messages])
+
+        expect(rows.first['messages'].map { |m| m['id'] }).to eq(%w[msg-1])
+        expect(WebMock).to have_requested(:get, "#{base}/issues/i1/messages").once
+      end
+
+      it 'reads the thread when the projection reaches inside it' do
+        stub_messages('i1', message_payload('msg-1'))
+
+        expect(issues.list(nil, filter, %w[id messages:body_html]).first['messages']).not_to be_nil
+      end
+
+      it 'reads no thread when the projection does not name it' do
+        rows = issues.list(nil, filter, %w[id title])
+
+        expect(rows.first).not_to have_key('messages')
+        expect(WebMock).not_to have_requested(:get, "#{base}/issues/i1/messages")
+      end
+
+      # A count and an export both go through a nil projection: neither asked
+      # for the conversation, and both would pay one request per row for it.
+      it 'reads no thread when the projection is nil' do
+        expect(issues.list(nil, filter, nil).first).not_to have_key('messages')
+        expect(WebMock).not_to have_requested(:get, "#{base}/issues/i1/messages")
+      end
+    end
+
+    describe 'the shape of an embedded message' do
+      before { stub_issues(issue_payload('i1')) }
+
+      it 'renames message_html and timestamp after the columns of the collection' do
+        stub_messages('i1', message_payload('msg-1'))
+
+        message = issues.list(nil, filter, %w[id messages]).first['messages'].first
+
+        expect(message).to include('id' => 'msg-1', 'body_html' => '<p>hello</p>', 'is_private' => false,
+                                   'source' => 'email', 'created_at' => '2026-08-07T13:06:22Z')
+      end
+
+      it 'fills every field the column schema declares' do
+        stub_messages('i1', message_payload('msg-1'))
+
+        message = issues.list(nil, filter, %w[id messages]).first['messages'].first
+
+        expect(message.keys).to match_array(Collections::Issue::MESSAGE_THREAD_SCHEMA.keys)
+      end
+
+      it 'keeps the thread in the order Pylon returns it, oldest first' do
+        stub_messages('i1', message_payload('msg-1'), message_payload('msg-2'), message_payload('msg-3'))
+
+        rows = issues.list(nil, filter, %w[id messages])
+
+        expect(rows.first['messages'].map { |m| m['id'] }).to eq(%w[msg-1 msg-2 msg-3])
+      end
+
+      it 'embeds an empty thread as an empty list' do
+        stub_messages('i1')
+
+        expect(issues.list(nil, filter, %w[id messages]).first['messages']).to eq([])
+      end
+    end
+
+    describe 'the author of a message' do
+      before { stub_issues(issue_payload('i1')) }
+
+      def author_of(payload)
+        stub_messages('i1', payload)
+        issues.list(nil, filter, %w[id messages]).first['messages'].first
+      end
+
+      it 'flattens the contact side of a message written by a customer' do
+        expect(author_of(message_payload('msg-1'))).to include(
+          'author_name' => 'Ada', 'author_avatar_url' => 'https://cdn/ada.png',
+          'author_email' => 'ada@acme.com', 'author_contact_id' => 'con-1', 'author_user_id' => nil
+        )
+      end
+
+      it 'flattens the user side of a message written by an agent' do
+        author = { 'name' => 'Grace', 'avatar_url' => nil, 'contact' => nil,
+                   'user' => { 'id' => 'usr-1', 'email' => 'grace@support.io' } }
+
+        expect(author_of(message_payload('msg-1', 'author' => author))).to include(
+          'author_name' => 'Grace', 'author_email' => 'grace@support.io',
+          'author_contact_id' => nil, 'author_user_id' => 'usr-1'
+        )
+      end
+
+      # Pylon puts both sides side by side with nothing telling them apart; the
+      # contact is the one the customer-facing thread is about.
+      it 'prefers the contact email when both sides are there' do
+        author = { 'name' => 'Ada', 'contact' => { 'id' => 'con-1', 'email' => 'ada@acme.com' },
+                   'user' => { 'id' => 'usr-1', 'email' => 'ada@support.io' } }
+
+        expect(author_of(message_payload('msg-1', 'author' => author))).to include(
+          'author_email' => 'ada@acme.com', 'author_contact_id' => 'con-1', 'author_user_id' => 'usr-1'
+        )
+      end
+
+      it 'leaves every author field null when the message carries no author' do
+        expect(author_of(message_payload('msg-1', 'author' => nil))).to include(
+          'author_name' => nil, 'author_email' => nil, 'author_avatar_url' => nil,
+          'author_contact_id' => nil, 'author_user_id' => nil
+        )
+      end
+    end
+
+    describe 'the fan-out it allows' do
+      let(:cap) { Collections::Issue::MAX_MESSAGE_EMBEDS }
+
+      before do
+        stub_issues(*(1..(cap + 2)).map { |i| issue_payload("i#{i}") })
+        (1..(cap + 2)).each { |i| stub_messages("i#{i}", message_payload("msg-#{i}")) }
+      end
+
+      it 'reads one thread per row up to the cap' do
+        issues.list(nil, filter, %w[id messages])
+
+        expect(WebMock).to have_requested(:get, "#{base}/issues/i#{cap}/messages").once
+        expect(WebMock).not_to have_requested(:get, "#{base}/issues/i#{cap + 1}/messages")
+      end
+
+      # Never the empty list: "no thread read" is not "this issue has no
+      # message", and the operator has to be able to tell them apart.
+      it 'leaves the rows past the cap at nil rather than at an empty thread' do
+        rows = issues.list(nil, filter, %w[id messages])
+
+        expect(rows.first(cap).map { |row| row['messages'] }).to all(be_an(Array))
+        expect(rows.drop(cap).map { |row| row['messages'] }).to eq([nil, nil])
+      end
+
+      it 'reports the rows it left out' do
+        allow(ForestAdminDatasourcePylon).to receive(:logger).and_return(logger)
+
+        issues.list(nil, filter, %w[id messages])
+
+        expect(logger).to have_received(:warn).with(/Asked for the message thread of #{cap + 2} issues/)
+      end
+
+      it 'reports nothing when the page fits under the cap' do
+        allow(ForestAdminDatasourcePylon).to receive(:logger).and_return(logger)
+        stub_issues(issue_payload('i1'))
+
+        issues.list(nil, filter, %w[id messages])
+
+        expect(logger).not_to have_received(:warn)
+      end
+    end
+
+    describe 'when the thread cannot be read' do
+      before { stub_issues(issue_payload('i1'), issue_payload('i2')) }
+
+      it 'serves the page with the column left at nil' do
+        stub_request(:get, "#{base}/issues/i1/messages").to_return(json({ 'message' => 'boom' }, 500))
+        stub_messages('i2', message_payload('msg-2'))
+
+        rows = issues.list(nil, filter, %w[id messages])
+
+        expect(rows.map { |row| row['id'] }).to eq(%w[i1 i2])
+        expect(rows.first['messages']).to be_nil
+        expect(rows.last['messages'].map { |m| m['id'] }).to eq(%w[msg-2])
+      end
+
+      it 'reports the degradation' do
+        allow(ForestAdminDatasourcePylon).to receive(:logger).and_return(logger)
+        stub_request(:get, %r{/issues/i\d/messages}).to_return(json({ 'message' => 'boom' }, 500))
+
+        issues.list(nil, filter, %w[id messages])
+
+        expect(logger).to have_received(:warn).with(/fetch_issue_messages\(i1\) failed; degrading/)
+      end
+    end
+  end
+end