diff --git a/osgi-bundles/bundles/logger/html-sse-demo.md b/osgi-bundles/bundles/logger/html-sse-demo.md
index a092ff57..bab4d2e2 100644
--- a/osgi-bundles/bundles/logger/html-sse-demo.md
+++ b/osgi-bundles/bundles/logger/html-sse-demo.md
@@ -4,25 +4,28 @@
 2. Copy the JAR to `platform` plugin directory
 3. Copy HTML content below to `<any-directory>`
 4. Run this using `python3 -m http.server 3000 -d <any-directory>` or `npx serve -l 3000 <any-directory>`
-5. Open `http://localhost:3000`, fill in the Base URL and your Kill Bill username/password, then click **Connect**
+5. Open `http://localhost:3000`, fill in the Base URL and your Kill Bill username/password, optionally the tenant API key/secret and the filters, then click **Connect**
 
-**Auth note:** this plugin's `/plugins/killbill-osgi-logger/**` endpoint now requires HTTP Basic Auth
-(see `BasicAuthFilter`, registered on this plugin's Jooby app only — other plugins are unaffected). The
-native browser `EventSource` API cannot set custom request headers (no way to send `Authorization: Basic
-...`), so this demo bundles [EventSource/eventsource](https://github.com/EventSource/eventsource) (MIT
-license, v5.1.2), a browser-compatible `EventSource` polyfill/replacement built on `fetch`, which supports
-a `fetch` override — used here to inject the Basic Auth header per request. It's bundled as a single
-minified IIFE (via `esbuild`, no build step required on your end) and pasted directly into the
-`<script>` tag below, so there's still nothing to install — just copy the HTML and go.
+### Filter combinations
 
-**CORS note:** this setup is cross-origin (demo on :3000, Kill Bill on :8080). Sending a custom
-`Authorization` header triggers a CORS preflight (`OPTIONS`); `BasicAuthFilter` explicitly lets
-`OPTIONS` through unauthenticated for this reason. If you serve the demo from the same origin as
-Kill Bill instead, the preflight doesn't happen at all.
+Filters combine with AND. Empty fields are ignored. Entries lacking the filtered attribute are excluded.
+`multitenant=false` servers skip the tenant checks (no `401`/`403`).
 
-**Filter note:** the `accountId`/`userToken` query-param filtering (`LogEntriesFilter`) is a separate,
-not-yet-merged feature (`platform-163`) and isn't wired up on this branch. The corresponding UI fields
-are kept below, commented out, so they're easy to restore once that feature lands here.
+| API key + secret | accountId | userToken | Result |
+|---|---|---|---|
+| - | - | - | Every entry, all tenants |
+| valid | - | - | Entries of that tenant |
+| valid | valid UUID, known | - | Entries of that tenant and account |
+| valid | valid UUID, unknown or other tenant's | - | Connected, heartbeats only |
+| - | - | token | Entries with that token, all tenants |
+| valid | - | token | Entries of that tenant with that token |
+| valid | valid UUID, known | token | Entries of that tenant and account with that token |
+| wrong, or only key or only secret | any | any | `401` |
+| - | any `accountId` | any | `403` |
+| valid | not a UUID | any | `400` |
+| any | `accountId` or `userToken` given twice | any | `400` |
+
+### HTML
 
 ```html
 <!DOCTYPE html>
@@ -70,17 +73,22 @@ are kept below, commented out, so they're easy to restore once that feature land
             <label>Password</label>
             <input id="password" type="password" placeholder="password" />
         </div>
-        <!--
-        TODO: re-enable once LogEntriesFilter / RecordIdApi (platform-163) lands on this branch.
         <div class="field">
-            <label>accountId (UUID)</label>
+            <label>X-Killbill-ApiKey (tenant)</label>
+            <input id="apiKey" placeholder="optional" />
+        </div>
+        <div class="field">
+            <label>X-Killbill-ApiSecret (tenant)</label>
+            <input id="apiSecret" type="password" placeholder="optional" />
+        </div>
+        <div class="field">
+            <label>accountId (UUID, needs API key/secret)</label>
             <input id="accountId" placeholder="optional" />
         </div>
         <div class="field">
             <label>userToken</label>
             <input id="userToken" placeholder="optional" />
         </div>
-        -->
         <button id="connectBtn">Connect</button>
         <button id="clearBtn">Clear</button>
         <span id="status" class="status closed">● Disconnected</span>
@@ -115,14 +123,12 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
 
         function buildUrl() {
             const base = document.getElementById('baseUrl').value.replace(/\/$/, '');
-            // TODO: re-enable once LogEntriesFilter / RecordIdApi (platform-163) lands on this branch.
-            // const accountId = document.getElementById('accountId').value.trim();
-            // const userToken = document.getElementById('userToken').value.trim();
-            // const params = [];
-            // if (accountId) params.push('accountId=' + encodeURIComponent(accountId));
-            // if (userToken) params.push('userToken=' + encodeURIComponent(userToken));
-            // return params.length > 0 ? base + '/?' + params.join('&') : base + '/';
-            return base + '/';
+            const accountId = document.getElementById('accountId').value.trim();
+            const userToken = document.getElementById('userToken').value.trim();
+            const params = [];
+            if (accountId) params.push('accountId=' + encodeURIComponent(accountId));
+            if (userToken) params.push('userToken=' + encodeURIComponent(userToken));
+            return params.length > 0 ? base + '/?' + params.join('&') : base + '/';
         }
 
         function buildAuthHeader() {
@@ -135,6 +141,17 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
             return 'Basic ' + utf8ToBase64(username + ':' + password);
         }
 
+        // Tenant headers, only when both are filled in; a half-filled pair is sent as is so the
+        // server's 401 for an incomplete pair can be observed.
+        function buildTenantHeaders() {
+            const apiKey = document.getElementById('apiKey').value.trim();
+            const apiSecret = document.getElementById('apiSecret').value.trim();
+            const headers = {};
+            if (apiKey) headers['X-Killbill-ApiKey'] = apiKey;
+            if (apiSecret) headers['X-Killbill-ApiSecret'] = apiSecret;
+            return headers;
+        }
+
         function utf8ToBase64(str) {
             const bytes = new TextEncoder().encode(str);
             let binary = '';
@@ -177,13 +194,15 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
 
             const url = buildUrl();
             const authHeader = buildAuthHeader();
+            const tenantHeaders = buildTenantHeaders();
             appendMessage('Connecting to: ' + url, 'msg-system');
 
-            // Custom fetch: injects the Basic Auth header on every (re)connect attempt, since the
-            // native EventSource API has no way to set request headers at all.
+            // Custom fetch: injects the Basic Auth and tenant headers on every (re)connect attempt,
+            // since the native EventSource API has no way to set request headers at all.
             const authFetch = function(input, init) {
                 const headers = new Headers((init && init.headers) || {});
                 if (authHeader) headers.set('Authorization', authHeader);
+                Object.keys(tenantHeaders).forEach(function(name) { headers.set(name, tenantHeaders[name]); });
                 return fetch(input, Object.assign({}, init, { headers: headers }));
             };
 
@@ -224,7 +243,14 @@ ${n}`,h++;break;case"id":n.includes("\0")||(f=n);break;case"retry":/^\d+$/.test(
                 // leave a stale instance quietly reconnecting in the background after we've already
                 // torn down the UI/state for it (which otherwise looks like "stuck/always connected"
                 // and makes the Disconnect button appear to do nothing).
-                const detail = event.code ? (' (HTTP ' + event.code + (event.message ? ': ' + event.message : '') + ')') : '';
+                const hints = {
+                    400: 'accountId is not a UUID, or accountId/userToken is repeated',
+                    401: 'bad Basic Auth credentials, or a wrong/incomplete API key + secret',
+                    403: 'accountId requires the API key + secret',
+                    429: 'too many failed attempts, retry later'
+                };
+                const hint = hints[event.code] ? ' - ' + hints[event.code] : '';
+                const detail = event.code ? (' (HTTP ' + event.code + (event.message ? ': ' + event.message : '') + hint + ')') : '';
                 appendMessage('Connection error' + detail, 'msg-error');
                 if (eventSource) {
                     eventSource.close();
diff --git a/osgi-bundles/bundles/logger/pom.xml b/osgi-bundles/bundles/logger/pom.xml
index a4223f78..cebfac7f 100644
--- a/osgi-bundles/bundles/logger/pom.xml
+++ b/osgi-bundles/bundles/logger/pom.xml
@@ -99,6 +99,17 @@
             <groupId>org.kill-bill.commons</groupId>
             <artifactId>killbill-utils</artifactId>
         </dependency>
+        <!-- Used only by TestLogsSseEndpoint, but it must stay compile scope: jooby already brings it in at compile scope,
+             so dependency-scope-maven-plugin rejects test scope and maven-dependency-plugin (analyze) requires it declared. -->
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-servlet</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java
index b54dc3b4..3df3a704 100644
--- a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java
+++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java
@@ -59,6 +59,9 @@ public void start(final BundleContext context) throws Exception {
         // Require HTTP Basic credentials (validated against Kill Bill RBAC), since /plugins/** is not authenticated
         // by the Kill Bill server by default. See BasicAuthFilter for details.
         pluginApp.use("*", new BasicAuthFilter(killbillAPI.getSecurityApi()));
+        // TenantFilter is only installed when multi-tenancy is enabled (the default, when the property is missing)
+        final boolean multiTenant = !"false".equalsIgnoreCase(configProperties.getString("org.killbill.server.multitenant"));
+        pluginApp.use("*", new LogsSseFilter(killbillAPI.getRecordIdApi(), multiTenant));
         logsSseHandler = new LogsSseHandler(logEntriesManager);
         pluginApp.sse("/", logsSseHandler);
         final HttpServlet httpServlet = PluginApp.createServlet(pluginApp);
diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseFilter.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseFilter.java
new file mode 100644
index 00000000..ae7d2c0e
--- /dev/null
+++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseFilter.java
@@ -0,0 +1,155 @@
+/*
+ * Copyright 2020-2026 Equinix, Inc
+ * Copyright 2014-2026 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.osgi.bundles.logger;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.jooby.Mutant;
+import org.jooby.Request;
+import org.jooby.Response;
+import org.jooby.Route;
+import org.killbill.billing.ObjectType;
+import org.killbill.billing.plugin.api.PluginTenantContext;
+import org.killbill.billing.tenant.api.Tenant;
+import org.killbill.billing.util.api.RecordIdApi;
+import org.killbill.billing.util.callcontext.TenantContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+// Validates the filtering parameters (accountId, userToken) and tenant headers, and resolves them once per connection
+// into a SseParamsCriteria stored on the request. This must be a filter (not done in the SSE handler): by the time the
+// handler runs, the 200 status and text/event-stream content type have already been sent, so errors can't be returned.
+//
+// Evaluation order (first failure answers): 401 (tenant headers not validated), 403 (accountId without tenant), 400 (syntax).
+class LogsSseFilter implements Route.Filter {
+
+    private static final Logger logger = LoggerFactory.getLogger(LogsSseFilter.class);
+
+    static final String TENANT_ATTRIBUTE = "killbill_tenant"; // Set by Kill Bill's TenantFilter after validating the API key/secret
+    static final String API_KEY_HEADER = "X-Killbill-ApiKey";
+    static final String API_SECRET_HEADER = "X-Killbill-ApiSecret";
+    static final String ACCOUNT_ID_PARAM = "accountId";
+    static final String USER_TOKEN_PARAM = "userToken";
+
+    private final RecordIdApi recordIdApi;
+    private final boolean multiTenant;
+
+    LogsSseFilter(final RecordIdApi recordIdApi, final boolean multiTenant) {
+        this.recordIdApi = recordIdApi;
+        this.multiTenant = multiTenant;
+    }
+
+    @Override
+    public void handle(final Request req, final Response rsp, final Route.Chain chain) throws Throwable {
+        // CORS preflight is answered by BasicAuthFilter and never reaches here, but be safe if the order changes
+        if ("OPTIONS".equalsIgnoreCase(req.method())) {
+            chain.next(req, rsp);
+            return;
+        }
+
+        final Tenant tenant = req.<Tenant>ifGet(TENANT_ATTRIBUTE).orElse(null);
+
+        // On plugin paths TenantFilter lets a request with a wrong or half-supplied key/secret through, without tenant
+        if (multiTenant && tenant == null && (req.header(API_KEY_HEADER).isSet() || req.header(API_SECRET_HEADER).isSet())) {
+            reject(rsp, 401, "Invalid " + API_KEY_HEADER + " / " + API_SECRET_HEADER);
+            return;
+        }
+
+        final List<String> accountIds = values(req, ACCOUNT_ID_PARAM);
+        final List<String> userTokens = values(req, USER_TOKEN_PARAM);
+        final String accountId = nonBlank(accountIds);
+        final String userToken = nonBlank(userTokens);
+
+        // accountId lookups are tenant-scoped
+        if (multiTenant && tenant == null && accountId != null) {
+            reject(rsp, 403, ACCOUNT_ID_PARAM + " requires " + API_KEY_HEADER + " / " + API_SECRET_HEADER);
+            return;
+        }
+
+        if (accountIds.size() > 1 || userTokens.size() > 1) {
+            reject(rsp, 400, "Parameters " + ACCOUNT_ID_PARAM + " and " + USER_TOKEN_PARAM + " can be specified only once");
+            return;
+        }
+
+        final UUID accountUuid;
+        if (accountId == null) {
+            accountUuid = null;
+        } else {
+            try {
+                accountUuid = UUID.fromString(accountId);
+            } catch (final IllegalArgumentException e) {
+                reject(rsp, 400, "Invalid " + ACCOUNT_ID_PARAM + ": not a UUID");
+                return;
+            }
+        }
+
+        final UUID tenantId = tenant == null ? null : tenant.getId();
+        String tenantRecordId = null;
+        if (tenantId != null) {
+            tenantRecordId = recordId(tenantId, ObjectType.TENANT, new PluginTenantContext(null, tenantId));
+            if (tenantRecordId == null) {
+                rejectUnresolvableTenant(rsp, tenantId);
+                return;
+            }
+        }
+
+        String accountRecordId = null;
+        if (accountUuid != null) {
+            accountRecordId = recordId(accountUuid, ObjectType.ACCOUNT, new PluginTenantContext(accountUuid, tenantId));
+        }
+
+        req.set(SseParamsCriteria.REQUEST_KEY, new SseParamsCriteria(tenantRecordId, accountUuid != null, accountRecordId, userToken));
+        chain.next(req, rsp);
+    }
+
+    // A tenant validated by TenantFilter must have a record id. If not, this is a server-side problem (not the client's),
+    // and continuing without tenant scoping would silently expose the entries of all tenants. Fail closed: answer 500
+    // before the stream opens, which also closes the connection.
+    private void rejectUnresolvableTenant(final Response rsp, final UUID tenantId) throws Throwable {
+        logger.error("Unable to resolve the record id of validated tenant {}, rejecting the log stream request", tenantId);
+        reject(rsp, 500, "Unable to resolve the tenant");
+    }
+
+    // Returns null when the object is unknown (or belongs to another tenant)
+    private String recordId(final UUID objectId, final ObjectType objectType, final TenantContext context) {
+        final Long recordId = recordIdApi.getRecordId(objectId, objectType, context);
+        if (recordId == null || recordId <= 0L) {
+            logger.debug("No record id for {} {}", objectType, objectId);
+            return null;
+        }
+        return recordId.toString();
+    }
+
+    private static List<String> values(final Request req, final String name) {
+        final Mutant param = req.param(name);
+        return param.isSet() ? param.toList(String.class) : List.of();
+    }
+
+    // A bare (?p), empty (?p=) or whitespace-only value means "not provided"
+    private static String nonBlank(final List<String> values) {
+        final Optional<String> first = values.stream().filter(v -> v != null && !v.isBlank()).findFirst();
+        return first.orElse(null);
+    }
+
+    private static void reject(final Response rsp, final int status, final String message) throws Throwable {
+        rsp.status(status);
+        rsp.send(message);
+    }
+}
diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java
index c2e90d36..c5d18f0e 100644
--- a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java
+++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java
@@ -29,7 +29,6 @@
 
 import org.jooby.Request;
 import org.jooby.Sse;
-import org.jooby.funzy.Throwing;
 import org.killbill.commons.concurrent.Executors;
 
 public class LogsSseHandler implements Sse.Handler, Closeable {
@@ -53,35 +52,38 @@ public void close() throws IOException {
 
     @Override
     public void handle(final Request req, final Sse sse) {
+        // Resolved (and validated) by LogsSseFilter
+        final SseParamsCriteria filter = req.<SseParamsCriteria>ifGet(SseParamsCriteria.REQUEST_KEY).orElse(SseParamsCriteria.NONE);
         final UUID lastEventId = sse.lastEventId(UUID.class).orElse(null);
         final UUID cacheId = UUID.fromString(sse.id());
         logEntriesManager.subscribe(cacheId, lastEventId);
 
         final AtomicReference<UUID> lastLogId = new AtomicReference<>();
-        final ScheduledFuture<?> future = scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
-            @Override
-            public void run() {
-                final Iterable<LogEntryJson> logEntries = logEntriesManager.drain(cacheId);
-                if (!logEntries.iterator().hasNext()) {
-                    // In case we have nothing to send, send a heartbeat to verify the client is still around
-                    // That way, we can more quickly cleanup our subscriptions
-                    // Note that we set the id as the last log id, so that we can easily resume
-                    sse.event("heartbeat").id(lastLogId.get()).send();
-                } else {
-                    for (final LogEntryJson logEntryJson : logEntries) {
-                        sse.event(logEntryJson).id(logEntryJson.getId()).send();
-                        lastLogId.set(logEntryJson.getId());
-                    }
+        final ScheduledFuture<?> future = scheduledExecutorService.scheduleAtFixedRate(() -> {
+            final Iterable<LogEntryJson> logEntries = filter.apply(logEntriesManager.drain(cacheId));
+            if (!logEntries.iterator().hasNext()) {
+                // In case we have nothing to send, send a heartbeat to verify the client is still around
+                // That way, we can more quickly cleanup our subscriptions
+                // Note that we set the id as the last log id, so that we can easily resume
+                // The id is omitted when no entry was sent yet (e.g. a filter matching nothing): Event#id rejects null,
+                // and the exception would silently cancel this periodic task
+                final Sse.Event heartbeat = sse.event("heartbeat");
+                final UUID lastId = lastLogId.get();
+                if (lastId != null) {
+                    heartbeat.id(lastId);
+                }
+                heartbeat.send();
+            } else {
+                for (final LogEntryJson logEntryJson : logEntries) {
+                    sse.event(logEntryJson).id(logEntryJson.getId()).send();
+                    lastLogId.set(logEntryJson.getId());
                 }
             }
         }, INITIAL_DELAY_MS, PERIOD_MS, TimeUnit.MILLISECONDS);
 
-        sse.onClose(new Throwing.Runnable() {
-            @Override
-            public void tryRun() throws Throwable {
-                future.cancel(true);
-                logEntriesManager.unsubscribe(cacheId);
-            }
+        sse.onClose(() -> {
+            future.cancel(true);
+            logEntriesManager.unsubscribe(cacheId);
         });
     }
 }
diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/SseParamsCriteria.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/SseParamsCriteria.java
new file mode 100644
index 00000000..b3912548
--- /dev/null
+++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/SseParamsCriteria.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2020-2026 Equinix, Inc
+ * Copyright 2014-2026 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.osgi.bundles.logger;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.annotation.Nullable;
+
+// Immutable, resolved once per SSE connection by LogsSseFilter. All active criteria are combined with AND, and an
+// entry lacking an attribute that is being filtered on is excluded.
+class SseParamsCriteria {
+
+    static final String REQUEST_KEY = SseParamsCriteria.class.getName();
+
+    static final SseParamsCriteria NONE = new SseParamsCriteria(null, false, null, null);
+
+    private final String tenantRecordId;
+    private final boolean accountRequested;
+    private final String accountRecordId;
+    private final String userToken;
+
+    // tenantRecordId: null means no tenant scoping.
+    // accountRequested: true when accountId was provided; accountRecordId null then means "no such account" (matches nothing).
+    // userToken: null means no token filtering.
+    SseParamsCriteria(@Nullable final String tenantRecordId,
+                     final boolean accountRequested,
+                     @Nullable final String accountRecordId,
+                     @Nullable final String userToken) {
+        this.tenantRecordId = tenantRecordId;
+        this.accountRequested = accountRequested;
+        this.accountRecordId = accountRecordId;
+        this.userToken = userToken;
+    }
+
+    boolean isActive() {
+        return tenantRecordId != null || accountRequested || userToken != null;
+    }
+
+    boolean match(final LogEntryJson entry) {
+        if (tenantRecordId != null && !tenantRecordId.equals(entry.getTenantRecordId())) {
+            return false;
+        }
+        if (accountRequested && (accountRecordId == null || !accountRecordId.equals(entry.getAccountRecordId()))) {
+            return false;
+        }
+        return userToken == null || userToken.equals(entry.getUserToken());
+    }
+
+    Iterable<LogEntryJson> apply(final Iterable<LogEntryJson> entries) {
+        if (!isActive()) {
+            return entries;
+        }
+        final List<LogEntryJson> result = new ArrayList<>();
+        for (final LogEntryJson entry : entries) {
+            if (match(entry)) {
+                result.add(entry);
+            }
+        }
+        return result;
+    }
+}
diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseEndpoint.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseEndpoint.java
new file mode 100644
index 00000000..699964d7
--- /dev/null
+++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseEndpoint.java
@@ -0,0 +1,248 @@
+/*
+ * Copyright 2020-2026 Equinix, Inc
+ * Copyright 2014-2026 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.osgi.bundles.logger;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.servlet.FilterHolder;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.killbill.billing.ObjectType;
+import org.killbill.billing.plugin.core.resources.jooby.PluginApp;
+import org.killbill.billing.plugin.core.resources.jooby.PluginAppBuilder;
+import org.killbill.billing.tenant.api.Tenant;
+import org.killbill.billing.util.api.RecordIdApi;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import jakarta.servlet.DispatcherType;
+import jakarta.servlet.Filter;
+import jakarta.servlet.http.HttpServletRequest;
+
+// Real Jetty + Jooby servlet + LogsSseFilter + LogsSseHandler. Only the tenant lookup is simulated:
+// a servlet filter plays the role of Kill Bill's TenantFilter and sets the "killbill_tenant" attribute for the valid API key.
+public class TestLogsSseEndpoint {
+
+    private static final String VALID_API_KEY = "good-key";
+    private static final int READ_TIMEOUT_MS = 5000;
+    private static final long DEADLINE_NANOS = 10_000_000_000L;
+    private static final String BASE_PATH = "/plugins/" + Activator.PLUGIN_NAME + "/";
+
+    private final UUID tenantId = UUID.randomUUID();
+    private final UUID accountId = UUID.randomUUID();
+    private final UUID otherAccountId = UUID.randomUUID();
+
+    private Server server;
+    private LogsSseHandler logsSseHandler;
+    private int port;
+
+    @BeforeClass(groups = "slow")
+    public void startServer() throws Exception {
+        final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class);
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(tenantId), Mockito.eq(ObjectType.TENANT), Mockito.any())).thenReturn(1L);
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())).thenReturn(10L);
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(otherAccountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())).thenReturn(11L);
+
+        final LogEntriesManager logEntriesManager = new LogEntriesManager();
+        logEntriesManager.recordEvent(new LogEntryJson(null, 3, "logger", "tenant1-account10", "tok-a", "1", "10", null));
+        logEntriesManager.recordEvent(new LogEntryJson(null, 3, "logger", "tenant1-account11", "tok-b", "1", "11", null));
+        logEntriesManager.recordEvent(new LogEntryJson(null, 3, "logger", "tenant2-account10", "tok-a", "2", "10", null));
+        logEntriesManager.recordEvent(new LogEntryJson(null, 3, "logger", "no-tenant", null, null, null, null));
+
+        logsSseHandler = new LogsSseHandler(logEntriesManager);
+        final PluginApp pluginApp = new PluginAppBuilder(Activator.PLUGIN_NAME).build();
+        pluginApp.use("*", new LogsSseFilter(recordIdApi, true));
+        pluginApp.sse("/", logsSseHandler);
+
+        final Tenant tenant = Mockito.mock(Tenant.class);
+        Mockito.when(tenant.getId()).thenReturn(tenantId);
+        final Filter tenantFilter = (request, response, chain) -> {
+            if (VALID_API_KEY.equals(((HttpServletRequest) request).getHeader(LogsSseFilter.API_KEY_HEADER))) {
+                request.setAttribute(LogsSseFilter.TENANT_ATTRIBUTE, tenant);
+            }
+            chain.doFilter(request, response);
+        };
+
+        server = new Server();
+        final ServerConnector connector = new ServerConnector(server);
+        server.addConnector(connector);
+        final ServletContextHandler context = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
+        context.addFilter(new FilterHolder(tenantFilter), "/*", java.util.EnumSet.of(DispatcherType.REQUEST));
+        final ServletHolder servlet = new ServletHolder(PluginApp.createServlet(pluginApp));
+        servlet.setAsyncSupported(true);
+        context.addServlet(servlet, "/*");
+        server.setHandler(context);
+        server.start();
+        port = connector.getLocalPort();
+    }
+
+    @AfterClass(groups = "slow", alwaysRun = true)
+    public void stopServer() throws Exception {
+        try {
+            if (server != null) {
+                server.stop();
+            }
+        } finally {
+            // Also shuts down the scheduler and closes the LogEntriesManager
+            if (logsSseHandler != null) {
+                logsSseHandler.close();
+            }
+        }
+    }
+
+    @Test(groups = "slow")
+    public void testNoTenantNoParamsStreamsEverything() throws Exception {
+        final Stream stream = open("", null);
+        Assert.assertEquals(stream.status, 200);
+        Assert.assertEquals(stream.awaitMessagesUntilHeartbeat(), List.of("tenant1-account10", "tenant1-account11", "tenant2-account10", "no-tenant"));
+    }
+
+    @Test(groups = "slow")
+    public void testValidTenantOnlySeesItsOwnEntries() throws Exception {
+        final Stream stream = open("", VALID_API_KEY);
+        Assert.assertEquals(stream.status, 200);
+        Assert.assertEquals(stream.awaitMessagesUntilHeartbeat(), List.of("tenant1-account10", "tenant1-account11"));
+    }
+
+    @Test(groups = "slow")
+    public void testAccountIdNarrowsWithinTenant() throws Exception {
+        final Stream stream = open("?accountId=" + accountId, VALID_API_KEY);
+        Assert.assertEquals(stream.status, 200);
+        Assert.assertEquals(stream.awaitMessagesUntilHeartbeat(), List.of("tenant1-account10"));
+    }
+
+    @Test(groups = "slow")
+    public void testUserTokenAndAccountIdAreAnded() throws Exception {
+        final Stream stream = open("?accountId=" + otherAccountId + "&userToken=tok-a", VALID_API_KEY);
+        Assert.assertEquals(stream.status, 200);
+        // Account 11 has only tok-b, so nothing can match: the connection stays open and only heartbeats arrive
+        Assert.assertEquals(stream.awaitHeartbeats(2), 2);
+        Assert.assertTrue(stream.messages.isEmpty());
+    }
+
+    @Test(groups = "slow")
+    public void testNoMatchOnlyHeartbeatsWithoutId() throws Exception {
+        final Stream stream = open("?userToken=nobody", VALID_API_KEY);
+        Assert.assertEquals(stream.status, 200);
+        Assert.assertEquals(stream.awaitHeartbeats(2), 2);
+        Assert.assertTrue(stream.messages.isEmpty());
+        Assert.assertTrue(stream.ids.isEmpty(), "Heartbeat before any entry must not carry an id");
+    }
+
+    @Test(groups = "slow")
+    public void testAccountIdWithoutTenantIs403() throws Exception {
+        Assert.assertEquals(open("?accountId=" + accountId, null).status, 403);
+    }
+
+    @Test(groups = "slow")
+    public void testApiKeyHeaderWithoutValidTenantIs401() throws Exception {
+        Assert.assertEquals(open("", "bad-key").status, 401);
+    }
+
+    @Test(groups = "slow")
+    public void testMalformedAccountIdIs400() throws Exception {
+        Assert.assertEquals(open("?accountId=not-a-uuid", VALID_API_KEY).status, 400);
+    }
+
+    @Test(groups = "slow")
+    public void testRepeatedParameterIs400() throws Exception {
+        Assert.assertEquals(open("?userToken=a&userToken=b", VALID_API_KEY).status, 400);
+        Assert.assertEquals(open("?accountId=" + accountId + "&accountId=" + otherAccountId, VALID_API_KEY).status, 400);
+    }
+
+    private Stream open(final String query, final String apiKey) throws IOException {
+        final HttpURLConnection conn = (HttpURLConnection) new URL("http://localhost:" + port + BASE_PATH + query).openConnection();
+        conn.setReadTimeout(READ_TIMEOUT_MS);
+        conn.setRequestProperty("Accept", "text/event-stream");
+        if (apiKey != null) {
+            conn.setRequestProperty(LogsSseFilter.API_KEY_HEADER, apiKey);
+            conn.setRequestProperty(LogsSseFilter.API_SECRET_HEADER, "secret");
+        }
+        return new Stream(conn);
+    }
+
+    private static final class Stream {
+
+        private final HttpURLConnection conn;
+        private final int status;
+        private final List<String> messages = new ArrayList<>();
+        private final List<String> ids = new ArrayList<>();
+        private final List<String> raw = new ArrayList<>();
+        private BufferedReader reader;
+
+        Stream(final HttpURLConnection conn) throws IOException {
+            this.conn = conn;
+            this.status = conn.getResponseCode();
+            if (status == 200) {
+                this.reader = new BufferedReader(new InputStreamReader(conn.getInputStream(), StandardCharsets.UTF_8));
+            }
+        }
+
+        // The handler sends a heartbeat only once the replayed entries are drained, so reading up to it sees every entry that would be sent
+        List<String> awaitMessagesUntilHeartbeat() throws IOException {
+            awaitHeartbeats(1);
+            return new ArrayList<>(messages);
+        }
+
+        int awaitHeartbeats(final int expected) throws IOException {
+            final long deadline = System.nanoTime() + DEADLINE_NANOS;
+            int heartbeats = 0;
+            try {
+                String line;
+                while (heartbeats < expected && System.nanoTime() < deadline && (line = reader.readLine()) != null) {
+                    raw.add(line);
+                    if (line.equals("data:heartbeat") || line.equals("data:\"heartbeat\"")) {
+                        heartbeats++;
+                    } else {
+                        consume(line);
+                    }
+                }
+                Assert.assertTrue(heartbeats >= expected, "Timed out. Heartbeats=" + heartbeats + " raw lines: " + raw);
+                return heartbeats;
+            } finally {
+                conn.disconnect();
+            }
+        }
+
+        private void consume(final String line) {
+            if (line.startsWith("id:")) {
+                ids.add(line.substring(3));
+            } else if (line.startsWith("data:{")) {
+                final String json = line.substring(5);
+                final int idx = json.indexOf("\"message\":\"");
+                if (idx >= 0) {
+                    final int start = idx + "\"message\":\"".length();
+                    messages.add(json.substring(start, json.indexOf('"', start)));
+                }
+            }
+        }
+    }
+}
diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseFilter.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseFilter.java
new file mode 100644
index 00000000..2d18b70d
--- /dev/null
+++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseFilter.java
@@ -0,0 +1,231 @@
+/*
+ * Copyright 2020-2026 Equinix, Inc
+ * Copyright 2014-2026 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.osgi.bundles.logger;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+
+import org.jooby.Mutant;
+import org.jooby.Request;
+import org.jooby.Response;
+import org.jooby.Route;
+import org.killbill.billing.ObjectType;
+import org.killbill.billing.tenant.api.Tenant;
+import org.killbill.billing.util.api.RecordIdApi;
+import org.killbill.billing.util.callcontext.TenantContext;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class TestLogsSseFilter {
+
+    private final UUID tenantId = UUID.randomUUID();
+    private final UUID accountId = UUID.randomUUID();
+
+    private RecordIdApi recordIdApi;
+    private Response response;
+    private Route.Chain chain;
+
+    @BeforeMethod(groups = "fast")
+    public void setUp() {
+        recordIdApi = Mockito.mock(RecordIdApi.class);
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(tenantId), Mockito.eq(ObjectType.TENANT), Mockito.any())).thenReturn(5L);
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())).thenReturn(42L);
+        response = Mockito.mock(Response.class);
+        Mockito.when(response.status(Mockito.anyInt())).thenReturn(response);
+        chain = Mockito.mock(Route.Chain.class);
+    }
+
+    @Test(groups = "fast")
+    public void testNoTenantNoParamsStreamsEverything() throws Throwable {
+        final Request req = request(null, false, false);
+        run(true, req);
+        assertPassed(req, null);
+        Mockito.verifyNoInteractions(recordIdApi);
+    }
+
+    @Test(groups = "fast")
+    public void testValidTenantScopesByTenantRecordId() throws Throwable {
+        final Request req = request(tenant(), true, true);
+        run(true, req);
+        final SseParamsCriteria filter = assertPassed(req, null);
+        Assert.assertTrue(filter.match(entry("5", null, null)));
+        Assert.assertFalse(filter.match(entry("6", null, null)));
+    }
+
+    @Test(groups = "fast")
+    public void testApiHeadersWithoutTenantAttributeIs401() throws Throwable {
+        run(true, request(null, true, true));
+        assertRejected(401);
+        run(true, request(null, true, false));
+        run(true, request(null, false, true));
+        Mockito.verify(response, Mockito.times(3)).status(401);
+    }
+
+    @Test(groups = "fast")
+    public void testAccountIdWithoutTenantIs403() throws Throwable {
+        run(true, request(null, false, false, "accountId", accountId.toString()));
+        assertRejected(403);
+    }
+
+    @Test(groups = "fast")
+    public void testAccountIdFilterUsesTenantContext() throws Throwable {
+        final Request req = request(tenant(), true, true, "accountId", accountId.toString());
+        run(true, req);
+        final SseParamsCriteria filter = assertPassed(req, null);
+        Assert.assertTrue(filter.match(entry("5", "42", null)));
+        Assert.assertFalse(filter.match(entry("5", "43", null)));
+        Assert.assertFalse(filter.match(entry("6", "42", null)));
+
+        final ArgumentCaptor<TenantContext> ctx = ArgumentCaptor.forClass(TenantContext.class);
+        Mockito.verify(recordIdApi).getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), ctx.capture());
+        Assert.assertEquals(ctx.getValue().getTenantId(), tenantId);
+    }
+
+    @Test(groups = "fast")
+    public void testUnknownAccountStaysConnectedAndMatchesNothing() throws Throwable {
+        final UUID unknown = UUID.randomUUID();
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(unknown), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())).thenReturn(null);
+        final Request req = request(tenant(), true, true, "accountId", unknown.toString());
+        run(true, req);
+        final SseParamsCriteria filter = assertPassed(req, null);
+        Assert.assertFalse(filter.match(entry("5", "42", null)));
+        Assert.assertFalse(filter.match(entry("5", null, null)));
+    }
+
+    @Test(groups = "fast")
+    public void testMalformedAccountIdIs400() throws Throwable {
+        run(true, request(tenant(), true, true, "accountId", "not-a-uuid"));
+        assertRejected(400);
+    }
+
+    @Test(groups = "fast")
+    public void testRepeatedParametersAre400() throws Throwable {
+        run(true, request(tenant(), true, true, "accountId", accountId.toString(), accountId.toString()));
+        run(true, request(tenant(), true, true, "userToken", "a", "b"));
+        Mockito.verify(response, Mockito.times(2)).status(400);
+        Mockito.verify(chain, Mockito.never()).next(Mockito.any(Request.class), Mockito.any(Response.class));
+    }
+
+    @Test(groups = "fast")
+    public void testBlankParametersMeanNotProvided() throws Throwable {
+        // Even without a tenant: a blank accountId must not trigger the 403
+        for (final String blank : new String[]{"", " ", "  "}) {
+            final Request req = request(null, false, false, "accountId", blank);
+            run(true, req);
+            assertPassed(req, null);
+        }
+        final Request req = request(null, false, false, "userToken", "");
+        run(true, req);
+        final SseParamsCriteria filter = assertPassed(req, null);
+        Assert.assertFalse(filter.isActive());
+    }
+
+    @Test(groups = "fast")
+    public void testUserTokenIsPlainStringAndNeedsNoTenant() throws Throwable {
+        final Request req = request(null, false, false, "userToken", "not a uuid");
+        run(true, req);
+        final SseParamsCriteria filter = assertPassed(req, null);
+        Assert.assertTrue(filter.match(entry(null, null, "not a uuid")));
+        Assert.assertFalse(filter.match(entry(null, null, "other")));
+    }
+
+    @Test(groups = "fast")
+    public void testSingleTenantModeSkipsTenantChecks() throws Throwable {
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())).thenReturn(42L);
+        final Request req = request(null, true, true, "accountId", accountId.toString());
+        run(false, req);
+        final SseParamsCriteria filter = assertPassed(req, null);
+        Assert.assertTrue(filter.match(entry("0", "42", null)));
+
+        final ArgumentCaptor<TenantContext> ctx = ArgumentCaptor.forClass(TenantContext.class);
+        Mockito.verify(recordIdApi).getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), ctx.capture());
+        Assert.assertNull(ctx.getValue().getTenantId());
+    }
+
+    @Test(groups = "fast")
+    public void testUnresolvableTenantIs500AndNeverStreams() throws Throwable {
+        Mockito.when(recordIdApi.getRecordId(Mockito.eq(tenantId), Mockito.eq(ObjectType.TENANT), Mockito.any())).thenReturn(null);
+        run(true, request(tenant(), true, true));
+        assertRejected(500);
+        Mockito.verify(recordIdApi, Mockito.never()).getRecordId(Mockito.any(), Mockito.eq(ObjectType.ACCOUNT), Mockito.any());
+    }
+
+    // -- helpers --
+
+    private void run(final boolean multiTenant, final Request req) throws Throwable {
+        new LogsSseFilter(recordIdApi, multiTenant).handle(req, response, chain);
+    }
+
+    private void assertRejected(final int status) throws Throwable {
+        Mockito.verify(response).status(status);
+        Mockito.verify(chain, Mockito.never()).next(Mockito.any(Request.class), Mockito.any(Response.class));
+    }
+
+    private SseParamsCriteria assertPassed(final Request req, final Object unused) throws Throwable {
+        Mockito.verify(response, Mockito.never()).status(Mockito.anyInt());
+        Mockito.verify(chain).next(req, response);
+        final ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+        Mockito.verify(req).set(Mockito.eq(SseParamsCriteria.REQUEST_KEY), captor.capture());
+        return (SseParamsCriteria) captor.getValue();
+    }
+
+    private Tenant tenant() {
+        final Tenant tenant = Mockito.mock(Tenant.class);
+        Mockito.when(tenant.getId()).thenReturn(tenantId);
+        return tenant;
+    }
+
+    private static LogEntryJson entry(final String tenant, final String account, final String token) {
+        return new LogEntryJson(null, 3, "logger", "msg", token, tenant, account, null);
+    }
+
+    private static Request request(final Tenant tenant, final boolean apiKey, final boolean apiSecret, final String param, final String... values) {
+        final Request req = request(tenant, apiKey, apiSecret);
+        final Mutant mutant = Mockito.mock(Mutant.class);
+        Mockito.when(mutant.isSet()).thenReturn(true);
+        Mockito.when(mutant.toList(String.class)).thenReturn(List.of(values));
+        Mockito.when(req.param(param)).thenReturn(mutant);
+        return req;
+    }
+
+    private static Request request(final Tenant tenant, final boolean apiKey, final boolean apiSecret) {
+        final Request req = Mockito.mock(Request.class);
+        Mockito.when(req.method()).thenReturn("GET");
+        Mockito.when(req.ifGet(LogsSseFilter.TENANT_ATTRIBUTE)).thenReturn(Optional.ofNullable(tenant));
+        final Mutant keyHeader = header(apiKey);
+        final Mutant secretHeader = header(apiSecret);
+        Mockito.when(req.header(LogsSseFilter.API_KEY_HEADER)).thenReturn(keyHeader);
+        Mockito.when(req.header(LogsSseFilter.API_SECRET_HEADER)).thenReturn(secretHeader);
+        for (final String name : new String[]{"accountId", "userToken"}) {
+            final Mutant absent = Mockito.mock(Mutant.class);
+            Mockito.when(absent.isSet()).thenReturn(false);
+            Mockito.when(req.param(name)).thenReturn(absent);
+        }
+        return req;
+    }
+
+    private static Mutant header(final boolean present) {
+        final Mutant mutant = Mockito.mock(Mutant.class);
+        Mockito.when(mutant.isSet()).thenReturn(present);
+        return mutant;
+    }
+}
diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseHandler.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseHandler.java
new file mode 100644
index 00000000..38d57529
--- /dev/null
+++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseHandler.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2020-2026 Equinix, Inc
+ * Copyright 2014-2026 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.osgi.bundles.logger;
+
+import java.util.Optional;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+
+import org.jooby.Request;
+import org.jooby.Sse;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.testng.annotations.Test;
+
+public class TestLogsSseHandler {
+
+    private static final int TIMEOUT_MS = 5000;
+
+    @Test(groups = "fast")
+    public void testHeartbeatWithoutIdWhenNothingWasSentYet() throws Exception {
+        final LogsSseHandler handler = new LogsSseHandler(new LogEntriesManager());
+        try {
+            // Active criteria that match nothing: only heartbeats can be sent, and no entry was ever sent, so there is no last id.
+            // Sse.Event#id rejects null, and the resulting exception would silently cancel the periodic task.
+            final SseParamsCriteria matchesNothing = new SseParamsCriteria(null, false, null, "no-such-token");
+            final Sse sse = mockSse(UUID.randomUUID());
+            final Sse.Event heartbeat = mockEvent();
+            Mockito.when(sse.event("heartbeat")).thenReturn(heartbeat);
+
+            handler.handle(mockRequest(matchesNothing), sse);
+
+            // Heartbeats keep coming: the periodic task was not cancelled by an exception
+            Mockito.verify(heartbeat, Mockito.timeout(TIMEOUT_MS).atLeast(2)).send();
+            Mockito.verify(heartbeat, Mockito.never()).id(Mockito.any());
+        } finally {
+            handler.close();
+        }
+    }
+
+    @Test(groups = "fast")
+    public void testHeartbeatCarriesTheLastSentEntryId() throws Exception {
+        final LogEntriesManager logEntriesManager = new LogEntriesManager();
+        final LogsSseHandler handler = new LogsSseHandler(logEntriesManager);
+        try {
+            final LogEntryJson matching = new LogEntryJson(null, 3, "logger", "msg", "tok", "1", "2", null);
+            logEntriesManager.recordEvent(matching);
+            logEntriesManager.recordEvent(new LogEntryJson(null, 3, "logger", "msg", "other", "1", "2", null));
+
+            final Sse sse = mockSse(UUID.randomUUID());
+            final Sse.Event entryEvent = mockEvent();
+            final Sse.Event heartbeat = mockEvent();
+            Mockito.when(sse.event("heartbeat")).thenReturn(heartbeat);
+            Mockito.when(sse.event(Mockito.any(LogEntryJson.class))).thenReturn(entryEvent);
+
+            handler.handle(mockRequest(new SseParamsCriteria(null, false, null, "tok")), sse);
+
+            // Only the matching entry is sent, then heartbeats resume from its id
+            Mockito.verify(heartbeat, Mockito.timeout(TIMEOUT_MS).atLeastOnce()).id(matching.getId());
+            final ArgumentCaptor<LogEntryJson> sent = ArgumentCaptor.forClass(LogEntryJson.class);
+            Mockito.verify(sse, Mockito.atLeastOnce()).event(sent.capture());
+            sent.getAllValues().forEach(e -> org.testng.Assert.assertEquals(e.getUserToken(), "tok"));
+            Mockito.verify(entryEvent, Mockito.times(1)).send();
+        } finally {
+            handler.close();
+        }
+    }
+
+    @Test(groups = "fast")
+    public void testNoFilterOnTheRequestStreamsEverything() throws Exception {
+        final LogEntriesManager logEntriesManager = new LogEntriesManager();
+        final LogsSseHandler handler = new LogsSseHandler(logEntriesManager);
+        try {
+            logEntriesManager.recordEvent(new LogEntryJson(null, 3, "logger", "msg", null, null, null, null));
+            final Sse sse = mockSse(UUID.randomUUID());
+            final Sse.Event entryEvent = mockEvent();
+            Mockito.when(sse.event(Mockito.any(LogEntryJson.class))).thenReturn(entryEvent);
+            final Request req = Mockito.mock(Request.class);
+            Mockito.when(req.ifGet(SseParamsCriteria.REQUEST_KEY)).thenReturn(Optional.empty());
+
+            handler.handle(req, sse);
+
+            Mockito.verify(entryEvent, Mockito.timeout(TIMEOUT_MS).times(1)).send();
+        } finally {
+            handler.close();
+        }
+    }
+
+    private static Request mockRequest(final SseParamsCriteria criteria) {
+        final Request req = Mockito.mock(Request.class);
+        Mockito.when(req.ifGet(SseParamsCriteria.REQUEST_KEY)).thenReturn(Optional.of(criteria));
+        return req;
+    }
+
+    private static Sse mockSse(final UUID id) {
+        final Sse sse = Mockito.mock(Sse.class);
+        Mockito.when(sse.id()).thenReturn(id.toString());
+        Mockito.when(sse.lastEventId(UUID.class)).thenReturn(Optional.empty());
+        return sse;
+    }
+
+    private static Sse.Event mockEvent() {
+        final Sse.Event event = Mockito.mock(Sse.Event.class);
+        Mockito.when(event.id(Mockito.any())).thenReturn(event);
+        Mockito.when(event.send()).thenReturn(new CompletableFuture<>());
+        return event;
+    }
+}
diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestSseParamsCriteria.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestSseParamsCriteria.java
new file mode 100644
index 00000000..cde47f84
--- /dev/null
+++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestSseParamsCriteria.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright 2020-2026 Equinix, Inc
+ * Copyright 2014-2026 The Billing Project, LLC
+ *
+ * The Billing Project licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License.  You may obtain a copy of the License at:
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.killbill.billing.osgi.bundles.logger;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.StreamSupport;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class TestSseParamsCriteria {
+
+    private static LogEntryJson entry(final String tenant, final String account, final String token) {
+        return new LogEntryJson(null, 3, "logger", "msg", token, tenant, account, null);
+    }
+
+    private static List<LogEntryJson> apply(final SseParamsCriteria filter, final LogEntryJson... entries) {
+        return StreamSupport.stream(filter.apply(List.of(entries)).spliterator(), false).toList();
+    }
+
+    @Test(groups = "fast")
+    public void testNoFilterKeepsEverythingIncludingEntriesWithoutIds() {
+        Assert.assertFalse(SseParamsCriteria.NONE.isActive());
+        final LogEntryJson noIds = entry(null, null, null);
+        Assert.assertEquals(apply(SseParamsCriteria.NONE, noIds, entry("1", "2", "t")).size(), 2);
+    }
+
+    @Test(groups = "fast")
+    public void testTenantScoping() {
+        final SseParamsCriteria filter = new SseParamsCriteria("1", false, null, null);
+        final LogEntryJson mine = entry("1", null, null);
+        Assert.assertEquals(apply(filter, mine, entry("2", null, null), entry(null, null, null)), List.of(mine));
+    }
+
+    @Test(groups = "fast")
+    public void testAccountMatchesExactRecordId() {
+        final SseParamsCriteria filter = new SseParamsCriteria("1", true, "10", null);
+        final LogEntryJson mine = entry("1", "10", null);
+        // "100" must not match "10", an entry without account is excluded, and so is the same account id in another tenant
+        Assert.assertEquals(apply(filter, mine, entry("1", "100", null), entry("1", null, null), entry("2", "10", null)), List.of(mine));
+    }
+
+    @Test(groups = "fast")
+    public void testUnknownAccountMatchesNothing() {
+        final SseParamsCriteria filter = new SseParamsCriteria("1", true, null, null);
+        Assert.assertTrue(filter.isActive());
+        Assert.assertTrue(apply(filter, entry("1", "10", null), entry("1", null, null)).isEmpty());
+    }
+
+    @Test(groups = "fast")
+    public void testUserTokenAlone() {
+        final String token = UUID.randomUUID().toString();
+        final SseParamsCriteria filter = new SseParamsCriteria(null, false, null, token);
+        final LogEntryJson mine = entry("7", null, token);
+        Assert.assertEquals(apply(filter, mine, entry("7", null, "other"), entry("7", null, null)), List.of(mine));
+    }
+
+    @Test(groups = "fast")
+    public void testFiltersCombineWithAnd() {
+        final SseParamsCriteria filter = new SseParamsCriteria("1", true, "10", "tok");
+        final LogEntryJson mine = entry("1", "10", "tok");
+        Assert.assertEquals(apply(filter, mine, entry("1", "10", "x"), entry("1", "11", "tok"), entry("2", "10", "tok")), List.of(mine));
+    }
+}
diff --git a/pom.xml b/pom.xml
index 9ec4fc39..e5f7974a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>org.kill-bill.billing</groupId>
         <artifactId>killbill-oss-parent</artifactId>
-        <version>0.147.31</version>
+        <version>0.147.35</version>
     </parent>
     <artifactId>killbill-platform</artifactId>
     <version>0.42.4-SNAPSHOT</version>