feat(webapp): Recently queued section on runs list + listEntriesForEnv helper

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

@@ -0,0 +1,51 @@
+import { DateTime } from "~/components/primitives/DateTime";
+import { Header3 } from "~/components/primitives/Headers";
+import { Paragraph } from "~/components/primitives/Paragraph";
+
+export type RecentlyQueuedEntry = {
+  runId: string;
+  status: "QUEUED" | "DRAINING" | "FAILED" | "DONE";
+  createdAt: string | Date;
+};
+
+// Runs the mollifier has buffered but the drainer hasn't yet materialised
+// into Postgres. Without this surface they're invisible to the dashboard
+// during the buffered window — the paginated runs list is PG-only. We
+// render a compact header section so operators can see in-flight buffered
+// entries at a glance while still scrolling the regular list below.
+export function RecentlyQueuedSection({ entries }: { entries: RecentlyQueuedEntry[] }) {
+  if (entries.length === 0) return null;
+
+  return (
+    <div className="border-b border-grid-dimmed bg-charcoal-850 px-3 py-3">
+      <Header3 className="mb-2">Recently queued ({entries.length})</Header3>
+      <Paragraph variant="extra-small/dimmed" className="mb-2">
+        Triggers accepted into the burst buffer. They&apos;ll appear in the list below once the
+        drainer materialises them.
+      </Paragraph>
+      <ul className="space-y-1 text-text-bright">
+        {entries.map((entry) => (
+          <li key={entry.runId} className="flex items-center gap-3 text-xs">
+            <span className="font-mono">{entry.runId}</span>
+            <span
+              className={
+                entry.status === "FAILED"
+                  ? "text-error"
+                  : entry.status === "DRAINING"
+                    ? "text-warning"
+                    : "text-text-dimmed"
+              }
+            >
+              {entry.status === "FAILED"
+                ? "Failed"
+                : entry.status === "DRAINING"
+                  ? "Draining"
+                  : "Queued"}
+            </span>
+            <DateTime date={entry.createdAt} />
+          </li>
+        ))}
+      </ul>
+    </div>
+  );
+}

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

@@ -41,6 +41,8 @@ import { useProject } from "~/hooks/useProject";
 import { useSearchParams } from "~/hooks/useSearchParam";
 import { useShortcutKeys } from "~/hooks/useShortcutKeys";
 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";
@@ -94,6 +96,15 @@ 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(() => [])
+    : [];
+
   // 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
   // the current task filter), and we don't want either bleeding into the saved
@@ -112,13 +123,18 @@ 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 } = useTypedLoaderData<typeof loader>();
+  const { data, rootOnlyDefault, filters, recentlyQueued } = useTypedLoaderData<typeof loader>();
   const { isConnected } = useDevPresence();
   const project = useProject();
   const environment = useEnvironment();
@@ -141,6 +157,7 @@ export default function Page() {
         </PageAccessories>
       </NavBar>
       <PageBody scrollable={false}>
+        <RecentlyQueuedSection entries={recentlyQueued} />
         <SelectedItemsProvider
           initialSelectedItems={[]}
           maxSelectedItemCount={BULK_ACTION_RUN_LIMIT}

packages/redis-worker/src/mollifier/buffer.test.ts

@@ -1025,3 +1025,80 @@ describe("MollifierBuffer envs set lifecycle", () => {
     },
   );
 });
+
+describe("MollifierBuffer.listEntriesForEnv", () => {
+  redisTest(
+    "returns up to maxCount entries from the queue without consuming them",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      const buffer = new MollifierBuffer({
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        entryTtlSeconds: 600,
+        logger: new Logger("test", "log"),
+      });
+
+      try {
+        await buffer.accept({ runId: "r1", envId: "env_a", orgId: "org_1", payload: "{}" });
+        await buffer.accept({ runId: "r2", envId: "env_a", orgId: "org_1", payload: "{}" });
+        await buffer.accept({ runId: "r3", envId: "env_a", orgId: "org_1", payload: "{}" });
+
+        const entries = await buffer.listEntriesForEnv("env_a", 2);
+        expect(entries).toHaveLength(2);
+        const runIds = entries.map((e) => e.runId);
+        expect(new Set(runIds).size).toBe(2);
+        for (const id of runIds) expect(["r1", "r2", "r3"]).toContain(id);
+
+        // Non-destructive: the drainer can still pop all three.
+        const popped: string[] = [];
+        for (let i = 0; i < 3; i++) {
+          const entry = await buffer.pop("env_a");
+          if (entry) popped.push(entry.runId);
+        }
+        expect(new Set(popped)).toEqual(new Set(["r1", "r2", "r3"]));
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest("returns empty array when env queue is empty", { timeout: 20_000 }, async ({ redisContainer }) => {
+    const buffer = new MollifierBuffer({
+      redisOptions: {
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+        password: redisContainer.getPassword(),
+      },
+      entryTtlSeconds: 600,
+      logger: new Logger("test", "log"),
+    });
+
+    try {
+      expect(await buffer.listEntriesForEnv("env_empty", 10)).toEqual([]);
+    } finally {
+      await buffer.close();
+    }
+  });
+
+  redisTest("maxCount <= 0 returns empty without hitting redis", { timeout: 20_000 }, async ({ redisContainer }) => {
+    const buffer = new MollifierBuffer({
+      redisOptions: {
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+        password: redisContainer.getPassword(),
+      },
+      entryTtlSeconds: 600,
+      logger: new Logger("test", "log"),
+    });
+
+    try {
+      expect(await buffer.listEntriesForEnv("env_a", 0)).toEqual([]);
+      expect(await buffer.listEntriesForEnv("env_a", -5)).toEqual([]);
+    } finally {
+      await buffer.close();
+    }
+  });
+});

packages/redis-worker/src/mollifier/buffer.ts

@@ -128,6 +128,23 @@ export class MollifierBuffer {
     return this.redis.smembers(`mollifier:org-envs:${orgId}`);
   }
 
+  // Read-only listing of currently-queued entries for a single env. Used by
+  // the dashboard's "Recently queued" surface — LRANGE is non-destructive,
+  // so the drainer still pops these entries in order. Returns up to
+  // `maxCount` entries (the most-recently-queued ones, since accept LPUSHes
+  // onto the head). Each entry hash is fetched separately; a `null` from
+  // getEntry (TTL expired between LRANGE and HGETALL) is skipped.
+  async listEntriesForEnv(envId: string, maxCount: number): Promise<BufferEntry[]> {
+    if (maxCount <= 0) return [];
+    const runIds = await this.redis.lrange(`mollifier:queue:${envId}`, 0, maxCount - 1);
+    const entries: BufferEntry[] = [];
+    for (const runId of runIds) {
+      const entry = await this.getEntry(runId);
+      if (entry) entries.push(entry);
+    }
+    return entries;
+  }
+
   async ack(runId: string): Promise<void> {
     await this.redis.del(`mollifier:entries:${runId}`);
   }