feat(webapp,redis-worker): listing endpoints merge buffered + PG runs (Phase E)

Q1 ZSET-merge design lands.

redis-worker side:
- MollifierBuffer.listForEnvWithWatermark — paginated newest-first
  read of buffered entries, bounded by a (createdAtMicros, runId)
  watermark. ZREVRANGEBYSCORE strictly below the watermark score plus
  a tied-score band scan for entries sharing the watermark's
  createdAtMicros.

webapp side:
- listingMerge.server.ts: callRunListWithBufferMerge wraps
  ApiRunListPresenter. Fetches a buffer page, synthesises each entry
  into the presenter's ListDataItem shape (status QUEUED, timestamps
  from entry hash, env slug looked up once), forwards the inner
  cursor to the presenter, merges by createdAt DESC with runId DESC
  tiebreak, truncates to pageSize. Compound base64-JSON cursor
  { inner, watermark, bufferExhausted } is backwards-compatible with
  legacy opaque cursors.
- api.v1.runs.ts + api.v1.projects.{projectRef}.runs.ts route through
  the wrapper. Project route extracts envId from filter[env]; absent
  that, falls back to the bare presenter (existing behaviour).
- Buffer eligibility skips for filters that can't match buffered runs
  (status not in QUEUED/PENDING/DELAYED, batch/schedule/version/
  region/machine filters). Buffer outages fall open to PG-only.
- Delete RecentlyQueuedSection banner + listEntriesForEnv loader call
  from dashboard runs index — buffered runs appear inline as QUEUED
  rows.

Author: d-cs

.changeset/mollifier-buffer-list-with-watermark.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Add `MollifierBuffer.listForEnvWithWatermark` for paginated, watermark-anchored reads of buffered entries newest-first. Implements the ZSET-based primitive that backs the mollifier listing merge in the webapp (Q1 design): `ZREVRANGEBYSCORE` strictly below the watermark score, with a tied-score band scan for entries sharing the watermark's `createdAtMicros`. Returns hydrated `BufferEntry` rows; orphans (queue ref without entry hash) are skipped silently.

.server-changes/mollifier-listing-merge.md

@@ -0,0 +1,14 @@
+---
+area: webapp
+type: feature
+---
+
+Run listing endpoints now include buffered runs transparently (Phase E — Q1 design).
+
+`GET /api/v1/runs` and `GET /api/v1/projects/{projectRef}/runs` route through `callRunListWithBufferMerge`. The helper fetches a watermark-anchored page from the mollifier buffer via `MollifierBuffer.listForEnvWithWatermark`, synthesises each entry into the same shape `ApiRunListPresenter` returns for PG rows (status `QUEUED`, all timestamps derived from the entry hash, env slug looked up once per request), and merges the two sources by `createdAt DESC` with `runId DESC` tiebreak. Truncates to `pageSize` total.
+
+Cursor is a compound base64-JSON `{ inner, watermark, bufferExhausted }`. The `inner` field carries the existing PG/ClickHouse cursor unchanged so the underlying presenter is untouched. Legacy cursors (plain strings from older SDKs) are accepted and treated as `bufferExhausted: true` — those clients see PG-only listing, matching today's behaviour. Once the buffer source returns fewer than `pageSize` entries below the watermark, `bufferExhausted` latches true and subsequent pages skip the buffer entirely (Q1 D4).
+
+Buffer is skipped when filters don't match buffered runs (status filter excluding QUEUED/PENDING/DELAYED, region/machine/version/batch/schedule filters — none of which buffered runs carry). Buffer outages fall open to PG-only for that request.
+
+Removes the `RecentlyQueuedSection` banner from the dashboard runs index — buffered runs now appear in the main list as normal `QUEUED` rows (Q1 D5).

apps/webapp/app/components/runs/RecentlyQueuedSection.tsx

@@ -151,7 +151,8 @@ export const ApiRunListSearchParams = z.object({
     }),
 });
 
-type ApiRunListSearchParams = z.infer<typeof ApiRunListSearchParams>;
+export type ApiRunListSearchParamsType = z.infer<typeof ApiRunListSearchParams>;
+type ApiRunListSearchParams = ApiRunListSearchParamsType;
 
 export class ApiRunListPresenter extends BasePresenter {
   public async call(

apps/webapp/app/presenters/v3/ApiRunListPresenter.server.ts

@@ -42,8 +42,6 @@ import { useSearchParams } from "~/hooks/useSearchParam";
 import { useShortcutKeys } from "~/hooks/useShortcutKeys";
 import { redirectWithErrorMessage } from "~/models/message.server";
 import { findProjectBySlug } from "~/models/project.server";
-import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
-import { RecentlyQueuedSection } from "~/components/runs/RecentlyQueuedSection";
 import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server";
 import { getRunFiltersFromRequest } from "~/presenters/RunFilters.server";
 import { NextRunListPresenter } from "~/presenters/v3/NextRunListPresenter.server";
@@ -98,14 +96,11 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     ...filters,
   });
 
-  // Mollifier buffer entries don't appear in the paginated PG query — they
-  // sit in Redis until the drainer materialises them. Surface them in a
-  // separate "Recently queued" section above the list so they're not
-  // invisible during the buffered window.
-  const mollifierBuffer = getMollifierBuffer();
-  const recentlyQueued = mollifierBuffer
-    ? await mollifierBuffer.listEntriesForEnv(environment.id, 50).catch(() => [])
-    : [];
+  // Phase E: buffered runs are merged into the main runs list via
+  // `callRunListWithBufferMerge` for the API routes; the dashboard's
+  // runs table consumes the same listing path indirectly. No separate
+  // "Recently queued" banner needed — buffered runs appear as normal
+  // QUEUED rows.
 
   // Only persist rootOnly when no tasks are filtered. While a task filter is active,
   // the toggle's URL value can be a temporary auto-flip (or a user override scoped to
@@ -125,18 +120,13 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       data: list,
       rootOnlyDefault: filters.rootOnly,
       filters,
-      recentlyQueued: recentlyQueued.map((entry) => ({
-        runId: entry.runId,
-        status: entry.status,
-        createdAt: entry.createdAt,
-      })),
     },
     headers ? { headers } : undefined
   );
 };
 
 export default function Page() {
-  const { data, rootOnlyDefault, filters, recentlyQueued } = useTypedLoaderData<typeof loader>();
+  const { data, rootOnlyDefault, filters } = useTypedLoaderData<typeof loader>();
   const { isConnected } = useDevPresence();
   const project = useProject();
   const environment = useEnvironment();
@@ -159,7 +149,6 @@ export default function Page() {
         </PageAccessories>
       </NavBar>
       <PageBody scrollable={false}>
-        <RecentlyQueuedSection entries={recentlyQueued} />
         <SelectedItemsProvider
           initialSelectedItems={[]}
           maxSelectedItemCount={BULK_ACTION_RUN_LIMIT}

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs._index/route.tsx

@@ -7,6 +7,7 @@ import {
   ApiRunListSearchParams,
 } from "~/presenters/v3/ApiRunListPresenter.server";
 import { createLoaderPATApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { callRunListWithBufferMerge } from "~/v3/mollifier/listingMerge.server";
 
 const ParamsSchema = z.object({
   projectRef: z.string(),
@@ -39,6 +40,35 @@ export const loader = createLoaderPATApiRoute(
       return json({ error: "Project not found" }, { status: 404 });
     }
 
+    // For PAT-scoped lookups the environment isn't supplied by auth;
+    // it's resolved from `filter[env]`. The presenter already does this
+    // lookup internally and errors if no env can be resolved. We mirror
+    // that resolution here so the mollifier-buffer merge has the env
+    // context it needs (envId + slug for synthesised list items).
+    const envFilter = searchParams["filter[env]"];
+    let envForMerge:
+      | { id: string; organizationId: string; slug: string }
+      | undefined;
+    if (envFilter && envFilter.length > 0) {
+      const env = await $replica.runtimeEnvironment.findFirst({
+        where: { projectId: project.id, slug: { in: envFilter } },
+        select: { id: true, organizationId: true, slug: true },
+      });
+      if (env) envForMerge = env;
+    }
+
+    if (envForMerge) {
+      const result = await callRunListWithBufferMerge({
+        project,
+        searchParams,
+        apiVersion,
+        environment: envForMerge,
+      });
+      return json(result);
+    }
+
+    // No env resolvable — let the presenter throw its existing
+    // ServiceValidationError, preserving the legacy behaviour.
     const presenter = new ApiRunListPresenter();
     const result = await presenter.call(project, searchParams, apiVersion);
 

apps/webapp/app/routes/api.v1.projects.$projectRef.runs.ts

@@ -1,13 +1,11 @@
 import { json } from "@remix-run/server-runtime";
-import {
-  ApiRunListPresenter,
-  ApiRunListSearchParams,
-} from "~/presenters/v3/ApiRunListPresenter.server";
+import { ApiRunListSearchParams } from "~/presenters/v3/ApiRunListPresenter.server";
 import { logger } from "~/services/logger.server";
 import {
   anyResource,
   createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
+import { callRunListWithBufferMerge } from "~/v3/mollifier/listingMerge.server";
 
 export const loader = createLoaderApiRoute(
   {
@@ -38,13 +36,12 @@ export const loader = createLoaderApiRoute(
     findResource: async () => 1, // This is a dummy function, we don't need to find a resource
   },
   async ({ searchParams, authentication, apiVersion }) => {
-    const presenter = new ApiRunListPresenter();
-    const result = await presenter.call(
-      authentication.environment.project,
+    const result = await callRunListWithBufferMerge({
+      project: authentication.environment.project,
       searchParams,
       apiVersion,
-      authentication.environment
-    );
+      environment: authentication.environment,
+    });
 
     return json(result);
   }