diff --git a/CHANGELOG.md b/CHANGELOG.md
index ab5332fd..82e3299b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,8 @@
 
 - **SSE delivery no longer strips newlines from broadcast payloads — multiline payloads are framed as consecutive `data:` lines per the SSE spec (issue #392).** `Streams::Envelope.message` collapsed `\r`/`\n` in the payload to nothing before writing the single `data:` line, silently corrupting any whitespace-significant broadcast (pre-formatted `<pre>` content, textarea seeds, JSON-in-data frames) on **both** the ephemeral and durable delivery paths — HTML's whitespace tolerance is why it went unnoticed. A multiline payload is now split on `\r\n`/`\r`/`\n` into consecutive `data:` lines, which EventSource clients rejoin with `\n`, making delivery lossless (a trailing newline survives via an empty final `data:` line; `\r` variants normalize to `\n` — SSE line terminators cannot be carried raw). The original injection defense is preserved: every payload line carries the `data:` prefix followed by one space, so a crafted payload still cannot forge `id:`/`event:` fields, and single-line fields (event names, comments) still strip newlines. The `<pgbus-stream-source>` element's fetch-path parser had the matching client-side bug — it joined `data:` lines without `\n` *and* `trim()`ed payload whitespace — and now follows EventSource semantics (join with `\n`, strip only the single leading space). Refs #392.
 
+- **Ephemeral broadcasts over the PG NOTIFY payload cap no longer fail — loudly on the sync path, silently in the coalescer — they auto-degrade to a durable publish (issue #391).** Ephemeral frames ride the NOTIFY payload itself, which PostgreSQL caps below 8000 bytes. Any rendered-component broadcast (a progress card with Tailwind classes easily exceeds it) previously raised `PGMQ::Errors::ConnectionError: … payload string too long` — an error class that sent diagnosis toward the connection, not the payload — and on the `coalesce:` path that raise happened inside the coalescer's flush thread, reaching no caller, no ErrorReporter, no log: small frames delivered, big frames vanished, and the operator saw "SSE works but updates don't arrive". Three changes: **(1)** `Stream#broadcast` now measures the wrapped JSON before the NOTIFY and publishes an over-budget frame durably instead (payload stored in PGMQ, the queue's insert trigger fires the NOTIFY as a bare wake on the same channel the subscriber already LISTENs on) — delivery semantics preserved on both the sync and coalesced paths, warn-logged and instrumented (`pgbus.stream.broadcast` with `ephemeral_fallback: true`). **(2)** Direct `Client#notify_stream` callers get publish-time validation: a typed `Pgbus::Streams::PayloadTooLarge` raised at the call site for payloads exceeding `Pgbus::Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES` (7999 bytes, the largest accepted payload), naming the stream, the byte count, and the durable-mode escape hatch. **(3)** The coalescer's flush thread routes every flush error through `ErrorReporter` (same report-don't-log reasoning as #352) — a background thread swallowing delivery failures is invisible to APM by construction. ⚠️ **Upgrade note for 0.13 installs:** `streams_default_broadcast_mode` defaults to `:ephemeral`, and that default is a **behavior change** for apps broadcasting rendered components (what `broadcast_render`-style usage produces) — before this fix, any frame over ~8KB was silently lost or misdiagnosed. Durable is the right mode for turbo-stream UI regardless (since-id replay needs the archive): pin `config.streams_default_broadcast_mode = :durable`, or use `streams_durable_patterns` for the streams that need it; the auto-fallback now covers whatever stays ephemeral. Refs #391.
+
 - **Queue-age metrics no longer count vt-parked (scheduled/retrying) messages — one delayed job stops reading as a degraded queue (issue #389).** ⚠️ **Behavior change on the AppSignal `pgbus_queue_latency` gauge.** pgmq's `oldest_msg_age_sec` is computed from `enqueued_at` and ignores `vt`, but a job enqueued with `wait:` or parked on a long retry backoff lives in the queue table with a future `vt` — that *is* the delayed-delivery mechanism. So a single parked message made the age metric grow at wall-clock rate for hours on an otherwise drained queue, and any latency alert thresholding on it fired continuously ("oldest message is 17045s old" on a healthy queue with depth 1, `read_ct` 0). Every metrics surface now also exposes **`oldest_claimable_age_sec`** — `now() - min(vt)` over rows with `vt <= now()`, i.e. the age of the oldest message actually *eligible for pickup*: an immediately-enqueued message contributes from enqueue time (matching the old number on a plain backlog), a scheduled/backoff-parked message contributes nothing until due, an in-flight message (vt pushed forward) is excluded, and nil means "no claimable backlog" even when the table is non-empty. Surfaces: `Web::DataSource` (dashboard, JSON API, MCP `pgbus_queues` tool), a new Prometheus gauge `pgbus_queue_oldest_claimable_age_seconds`, a new AppSignal gauge `pgbus_queue_oldest_claimable_age_seconds`, `Pgbus::Client#oldest_claimable_ages` (raw-SQL reader, since pgmq's `metrics_result` type is frozen upstream), and a CLAIMABLE column in `pgbus queues`. The AppSignal **`pgbus_queue_latency` gauge now derives from the claimable age** and always emits — `(claimable_age || 0) * 1000`, 0 = no claimable backlog — so existing latency alerts stop false-firing with no dashboard changes; the raw `pgbus_queue_oldest_message_age_seconds` gauge keeps its enqueue-time semantics everywhere. The dashboard queue tables additionally split depth into **Parked** (`depth − visible`) and show the claimable age in place of the raw age, so a queue holding only backoff retries reads visibly healthy. Refs #389.
 
 ### Added
diff --git a/lib/pgbus/client/notify_stream.rb b/lib/pgbus/client/notify_stream.rb
index 1f9087e0..5f11498d 100644
--- a/lib/pgbus/client/notify_stream.rb
+++ b/lib/pgbus/client/notify_stream.rb
@@ -14,15 +14,24 @@ class Client
     # no orphan tables.
     #
     # The payload is JSON-serialized into the NOTIFY's optional payload
-    # parameter (max 8000 bytes in Postgres). Broadcasts exceeding this
-    # limit will raise a PG::ProgramLimitExceeded error — callers needing
-    # large payloads should use durable mode (which inserts into PGMQ).
+    # parameter. Postgres caps NOTIFY payloads at < 8000 bytes; oversized
+    # payloads raise a typed Pgbus::Streams::PayloadTooLarge here, at the
+    # call site, instead of surfacing as a misleading
+    # PGMQ::Errors::ConnectionError ("payload string too long") from deep
+    # inside the driver (issue #391). Callers needing large payloads should
+    # use durable mode (which inserts into PGMQ).
     module NotifyStream
+      # PostgreSQL rejects NOTIFY payloads of 8000 bytes or more
+      # ("payload string too long"), so 7999 is the largest deliverable
+      # payload.
+      NOTIFY_PAYLOAD_LIMIT_BYTES = 7999
+
       def notify_stream(stream_name, payload)
         full_name = config.queue_name(stream_name)
         sanitized = QueueNameValidator.sanitize!(full_name)
         channel = "pgmq.q_#{sanitized}.INSERT"
         json = payload.is_a?(String) ? payload : JSON.generate(payload)
+        validate_notify_payload_size!(stream_name, json)
 
         Instrumentation.instrument("pgbus.stream.notify", stream: stream_name, bytes: json.bytesize) do
           with_stale_connection_retry do
@@ -34,6 +43,19 @@ def notify_stream(stream_name, payload)
           end
         end
       end
+
+      private
+
+      def validate_notify_payload_size!(stream_name, json)
+        return if json.bytesize <= NOTIFY_PAYLOAD_LIMIT_BYTES
+
+        raise Pgbus::Streams::PayloadTooLarge,
+              "Ephemeral broadcast on stream #{stream_name.inspect} is #{json.bytesize} bytes; " \
+              "PostgreSQL caps NOTIFY payloads at #{NOTIFY_PAYLOAD_LIMIT_BYTES} bytes. " \
+              "Use durable mode for large payloads (payload stored in PGMQ, NOTIFY as wake) — " \
+              "e.g. broadcast(..., durable: true), a streams_durable_patterns match, or " \
+              "streams_default_broadcast_mode = :durable."
+      end
     end
   end
 end
diff --git a/lib/pgbus/streams.rb b/lib/pgbus/streams.rb
index 056f2c47..1edfe144 100644
--- a/lib/pgbus/streams.rb
+++ b/lib/pgbus/streams.rb
@@ -15,6 +15,18 @@ module Streams
     # this specifically can rescue Pgbus::Streams::StreamNameTooLong.
     class StreamNameTooLong < ArgumentError; end
 
+    # Raised when an ephemeral broadcast's JSON payload exceeds PostgreSQL's
+    # NOTIFY payload budget (< 8000 bytes). Ephemeral frames ride the NOTIFY
+    # itself, so the cap is a hard PostgreSQL limit — durable mode (payload
+    # stored in PGMQ, NOTIFY as a bare wake) has no such cap.
+    #
+    # Stream#broadcast never raises this: an oversized ephemeral frame
+    # auto-degrades to a durable publish (issue #391). The error exists for
+    # direct Client#notify_stream callers, where the previous failure mode
+    # was a misleading PGMQ::Errors::ConnectionError ("payload string too
+    # long") that pointed diagnosis at the connection instead of the payload.
+    class PayloadTooLarge < Pgbus::Error; end
+
     # The default SSE `event:` name for a broadcast frame. Turbo's
     # StreamObserver consumes frames the client re-dispatches as the
     # `message` DOM event; the client maps this SSE event name to
@@ -261,11 +273,47 @@ def self.validate_name_length!(name, streamables)
 
       private
 
+      # Ephemeral frames ride the NOTIFY payload itself, which PostgreSQL
+      # caps below 8000 bytes. A frame over the cap auto-degrades to a
+      # durable publish (issue #391): payload stored in PGMQ, the queue's
+      # insert trigger fires the NOTIFY as a bare wake on the same channel
+      # the subscriber already LISTENs on — delivery semantics preserved,
+      # cap irrelevant. The JSON is generated once here and passed
+      # pre-serialized to notify_stream so the size check costs no extra
+      # allocation on the hot path.
       def broadcast_ephemeral(wrapped)
-        @client.notify_stream(@name, wrapped)
+        json = JSON.generate(wrapped)
+        return durable_fallback(wrapped, json.bytesize) if json.bytesize > Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES
+
+        @client.notify_stream(@name, json)
         nil
       end
 
+      # The durable degrade path for an oversized ephemeral frame. Stays
+      # fire-and-forget like the ephemeral path it replaces: no after_commit
+      # deferral (pg_notify runs on the PGMQ pool connection, outside the
+      # request's AR transaction, so the ephemeral path never deferred
+      # either). Returns the msg_id like any durable publish.
+      def durable_fallback(wrapped, bytes)
+        Pgbus.logger.warn do
+          "[Pgbus::Streams] ephemeral broadcast on #{@name.inspect} is #{bytes} bytes " \
+            "(NOTIFY cap is #{Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES}); " \
+            "publishing durably instead. Consider durable mode for this stream " \
+            "(streams_durable_patterns or durable: true) to skip this check."
+        end
+        ensure_queue!
+        instrument_payload = {
+          stream: @name,
+          visible_to: wrapped["visible_to"],
+          deferred: false,
+          bytes: wrapped["html"].bytesize,
+          ephemeral_fallback: true
+        }
+        Instrumentation.instrument("pgbus.stream.broadcast", instrument_payload) do
+          @client.send_stream_message(@name, wrapped)
+        end
+      end
+
       # Submits a frame to the process-wide coalescer instead of
       # broadcasting now. Requires a target (the dedupe key — there's no
       # way to last-write-win without one). The window is `coalesce` in ms
diff --git a/lib/pgbus/streams/coalescer.rb b/lib/pgbus/streams/coalescer.rb
index 908423c1..660faf0e 100644
--- a/lib/pgbus/streams/coalescer.rb
+++ b/lib/pgbus/streams/coalescer.rb
@@ -73,6 +73,12 @@ def flush_key(key, stream_name, target)
         return unless entry
 
         @flush.call(stream_name: stream_name, target: target, payload: entry.payload, opts: entry.opts)
+      rescue StandardError => e
+        # The flush runs on the scheduler's thread — a raise here reaches no
+        # caller, so a swallowed error is invisible to APM by construction
+        # (issue #391: an oversized ephemeral frame died here without a
+        # trace). Route through ErrorReporter so configured reporters see it.
+        ErrorReporter.report(e, { component: "streams.coalescer", stream: stream_name, target: target })
       end
 
       # Default scheduler backed by Concurrent::ScheduledTask. Kept as a
diff --git a/spec/pgbus/client/notify_stream_spec.rb b/spec/pgbus/client/notify_stream_spec.rb
index a9cf3907..850e7f96 100644
--- a/spec/pgbus/client/notify_stream_spec.rb
+++ b/spec/pgbus/client/notify_stream_spec.rb
@@ -69,6 +69,50 @@ def initialize(*args, **kwargs); end
       )
     end
 
+    context "when the payload exceeds the PG NOTIFY budget" do
+      let(:oversized) { { "html" => "<div>#{"x" * 9000}</div>" } }
+
+      it "raises a typed Pgbus::Streams::PayloadTooLarge at the call site" do
+        expect do
+          client.notify_stream("chat", oversized)
+        end.to raise_error(Pgbus::Streams::PayloadTooLarge, /\d+ bytes.*7999/m)
+      end
+
+      it "names the stream and suggests durable mode in the message" do
+        expect do
+          client.notify_stream("chat", oversized)
+        end.to raise_error(Pgbus::Streams::PayloadTooLarge, /chat.*durable/m)
+      end
+
+      it "does not attempt the NOTIFY" do
+        begin
+          client.notify_stream("chat", oversized)
+        rescue Pgbus::Streams::PayloadTooLarge
+          nil
+        end
+        expect(raw_conn).not_to have_received(:exec_params)
+      end
+
+      it "is a Pgbus::Error so blanket pgbus rescues keep working" do
+        expect(Pgbus::Streams::PayloadTooLarge.ancestors).to include(Pgbus::Error)
+      end
+
+      it "measures bytes, not characters (multibyte payloads)" do
+        # 4500 two-byte chars = 4500 chars but 9000 bytes of HTML.
+        multibyte = { "html" => "é" * 4500 }
+        expect do
+          client.notify_stream("chat", multibyte)
+        end.to raise_error(Pgbus::Streams::PayloadTooLarge)
+      end
+
+      it "accepts a payload exactly at the budget" do
+        # {"html":""} wrapper is 11 bytes; fill to exactly 7999.
+        at_limit = { "html" => "x" * (7999 - 11) }
+        expect { client.notify_stream("chat", at_limit) }.not_to raise_error
+        expect(raw_conn).to have_received(:exec_params)
+      end
+    end
+
     context "with stale pgmq connection recovery" do
       before do
         real_pgmq_connection_error
diff --git a/spec/pgbus/streams/coalescer_spec.rb b/spec/pgbus/streams/coalescer_spec.rb
index 96a5519b..47f42e15 100644
--- a/spec/pgbus/streams/coalescer_spec.rb
+++ b/spec/pgbus/streams/coalescer_spec.rb
@@ -112,4 +112,42 @@ def submit(stream: "chat", target: "cursor", payload: "x", window_ms: 50, **opts
     scheduler.run_all
     expect(flushed.map { |f| f[:target] }).to eq(["cursor"])
   end
+
+  # Issue #391: the flush runs on the scheduler's thread — a raise there
+  # never reaches any caller, so a swallowed error is invisible to APM by
+  # construction. Errors must route through ErrorReporter (same reasoning
+  # as the StreamApp report-don't-log fix in #352).
+  describe "flush errors" do
+    let(:boom) { StandardError.new("payload string too long") }
+    let(:flush) { ->(**) { raise boom } }
+
+    it "routes a raising flush through ErrorReporter with stream/target context" do
+      allow(Pgbus::ErrorReporter).to receive(:report)
+
+      submit(stream: "chat", target: "cursor", payload: "a")
+      scheduler.run_all
+
+      expect(Pgbus::ErrorReporter).to have_received(:report).with(
+        boom, hash_including(
+                component: "streams.coalescer", stream: "chat", target: "cursor"
+              )
+      )
+    end
+
+    it "does not re-raise out of the flush" do
+      allow(Pgbus::ErrorReporter).to receive(:report)
+      submit(payload: "a")
+      expect { scheduler.run_all }.not_to raise_error
+    end
+
+    it "keeps the key usable for the next window after a failed flush" do
+      allow(Pgbus::ErrorReporter).to receive(:report)
+      submit(payload: "a")
+      scheduler.run_all
+
+      # A new submit after the failed flush schedules a fresh window.
+      submit(payload: "b")
+      expect(scheduler.scheduled.size).to eq(1)
+    end
+  end
 end
diff --git a/spec/pgbus/streams/ephemeral_broadcast_spec.rb b/spec/pgbus/streams/ephemeral_broadcast_spec.rb
index d4c9450c..c50618c3 100644
--- a/spec/pgbus/streams/ephemeral_broadcast_spec.rb
+++ b/spec/pgbus/streams/ephemeral_broadcast_spec.rb
@@ -34,11 +34,11 @@
       expect(client).not_to have_received(:send_stream_message)
     end
 
-    it "sends a PG NOTIFY with the payload" do
+    it "sends a PG NOTIFY with the pre-serialized payload" do
       stream.broadcast("<turbo-stream>X</turbo-stream>")
       expect(client).to have_received(:notify_stream).with(
         "chat",
-        { "html" => "<turbo-stream>X</turbo-stream>" }
+        JSON.generate({ "html" => "<turbo-stream>X</turbo-stream>" })
       )
     end
 
@@ -46,7 +46,7 @@
       stream.broadcast("<turbo-stream>X</turbo-stream>", visible_to: :admin_only)
       expect(client).to have_received(:notify_stream).with(
         "chat",
-        { "html" => "<turbo-stream>X</turbo-stream>", "visible_to" => "admin_only" }
+        JSON.generate({ "html" => "<turbo-stream>X</turbo-stream>", "visible_to" => "admin_only" })
       )
     end
 
diff --git a/spec/pgbus/streams/ephemeral_overflow_spec.rb b/spec/pgbus/streams/ephemeral_overflow_spec.rb
new file mode 100644
index 00000000..f8c9e41e
--- /dev/null
+++ b/spec/pgbus/streams/ephemeral_overflow_spec.rb
@@ -0,0 +1,112 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+
+# Issue #391: an ephemeral broadcast whose JSON exceeds the PG NOTIFY
+# payload cap must not raise a misleading connection error (sync path) or
+# vanish silently (coalescer flush thread). It auto-degrades to a durable
+# publish: payload in PGMQ, the queue's insert trigger fires the NOTIFY as
+# a bare wake on the same channel the subscriber already LISTENs on.
+RSpec.describe Pgbus::Streams::Stream do
+  subject(:stream) { described_class.new("probe", client: client, durable: false) }
+
+  let(:client) do
+    instance_double(
+      Pgbus::Client,
+      ensure_stream_queue: nil,
+      send_stream_message: 1248,
+      notify_stream: nil,
+      config: config
+    )
+  end
+
+  let(:config) do
+    Pgbus::Configuration.new.tap do |c|
+      c.queue_prefix = "pgbus_test"
+    end
+  end
+
+  let(:limit) { Pgbus::Client::NotifyStream::NOTIFY_PAYLOAD_LIMIT_BYTES }
+  let(:big_html) { "<div>#{"x" * 9000}</div>" }
+
+  describe "oversized ephemeral broadcast" do
+    it "falls back to a durable publish" do
+      stream.broadcast(big_html)
+      expect(client).to have_received(:send_stream_message).with("probe", { "html" => big_html })
+    end
+
+    it "ensures the stream queue before publishing" do
+      stream.broadcast(big_html)
+      expect(client).to have_received(:ensure_stream_queue).with("probe")
+    end
+
+    it "does not attempt the NOTIFY" do
+      stream.broadcast(big_html)
+      expect(client).not_to have_received(:notify_stream)
+    end
+
+    it "returns the durable msg_id" do
+      expect(stream.broadcast(big_html)).to eq(1248)
+    end
+
+    it "warn-logs the fallback with stream and byte count" do
+      warning = nil
+      allow(Pgbus.logger).to receive(:warn) { |&block| warning = block.call }
+      stream.broadcast(big_html)
+      expect(warning).to include(
+        '"probe"',
+        "#{JSON.generate({ "html" => big_html }).bytesize} bytes"
+      )
+    end
+
+    it "carries visible_to through to the durable publish" do
+      stream.broadcast(big_html, visible_to: :admins)
+      expect(client).to have_received(:send_stream_message).with(
+        "probe", { "html" => big_html, "visible_to" => "admins" }
+      )
+    end
+
+    it "instruments the fallback" do
+      events = []
+      allow(Pgbus::Instrumentation).to receive(:instrument) do |name, payload, &block|
+        events << [name, payload]
+        block&.call
+      end
+
+      stream.broadcast(big_html)
+
+      event = events.find { |(name, _)| name == "pgbus.stream.broadcast" }
+      expect(event).not_to be_nil
+      expect(event.last).to include(stream: "probe", ephemeral_fallback: true)
+    end
+  end
+
+  describe "budget boundary" do
+    it "measures the wrapped JSON, so metadata pushes a near-cap frame over" do
+      # HTML sized so {"html":"..."} is exactly at the limit; adding
+      # visible_to overflows the JSON and must trigger the fallback.
+      at_limit_html = "x" * (limit - 11)
+      stream.broadcast(at_limit_html, visible_to: :admins)
+      expect(client).to have_received(:send_stream_message)
+      expect(client).not_to have_received(:notify_stream)
+    end
+
+    it "keeps a frame exactly at the cap on the NOTIFY path" do
+      at_limit_html = "x" * (limit - 11)
+      stream.broadcast(at_limit_html)
+      expect(client).to have_received(:notify_stream)
+      expect(client).not_to have_received(:send_stream_message)
+    end
+  end
+
+  describe "small ephemeral broadcast" do
+    it "stays on the NOTIFY path, passing the pre-serialized JSON" do
+      stream.broadcast("<div>hi</div>")
+      expect(client).to have_received(:notify_stream).with(
+        "probe", JSON.generate({ "html" => "<div>hi</div>" })
+      )
+      expect(client).not_to have_received(:send_stream_message)
+      expect(client).not_to have_received(:ensure_stream_queue)
+    end
+  end
+end
diff --git a/spec/pgbus/streams_spec.rb b/spec/pgbus/streams_spec.rb
index acfdfd0f..6a1d1845 100644
--- a/spec/pgbus/streams_spec.rb
+++ b/spec/pgbus/streams_spec.rb
@@ -210,6 +210,23 @@ def run_callbacks! = @callbacks.each(&:call)
         # No run_callbacks! → simulates rollback.
         expect(coalescer).not_to have_received(:submit)
       end
+
+      # Pins the ephemeral half of the gating contract (issue #391 Q&A):
+      # transaction gating is decided at SUBMIT time by the requested mode.
+      # An ephemeral coalesced frame is fire-and-forget and submits
+      # immediately even inside an open transaction — so the oversized-frame
+      # durable fallback (which happens later, at flush, on the coalescer
+      # thread where no request transaction is visible) inherits exactly the
+      # ephemeral contract the caller chose, never a surprise deferral.
+      it "submits an ephemeral coalesced frame immediately, even inside the transaction" do
+        ephemeral = described_class.new("chat", client: client, durable: false)
+        ephemeral.broadcast("x", coalesce: 50, target: "t")
+
+        expect(coalescer).to have_received(:submit).with(
+          hash_including(target: "t", opts: hash_including(durable: false))
+        )
+        expect(transaction.callbacks).to be_empty
+      end
     end
 
     describe "#broadcast_render" do