From 5a7a12de4ee509ae0157fd9678b6e95ed0cce13b Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Thu, 30 Jul 2026 05:14:56 +0000 Subject: [PATCH] feat(core): paginate progress run listings --- .changeset/calm-runs-page.md | 5 ++++ packages/core/src/progress/progress.spec.ts | 28 ++++++++++++++++++ packages/core/src/progress/routes.ts | 30 ++++++++++++++++++- packages/core/src/progress/store.spec.ts | 32 +++++++++++++++++++++ packages/core/src/progress/store.ts | 19 +++++++++--- packages/core/src/progress/types.ts | 5 ++++ 6 files changed, 114 insertions(+), 5 deletions(-) create mode 100644 .changeset/calm-runs-page.md diff --git a/.changeset/calm-runs-page.md b/.changeset/calm-runs-page.md new file mode 100644 index 0000000000..88539a10af --- /dev/null +++ b/.changeset/calm-runs-page.md @@ -0,0 +1,5 @@ +--- +"@agent-native/core": minor +--- + +Add keyset pagination to progress-run listing through `ListRunsOptions.before` and the `beforeStartedAt`/`beforeId` HTTP query parameters. diff --git a/packages/core/src/progress/progress.spec.ts b/packages/core/src/progress/progress.spec.ts index 8de000fc7b..f9677ea39a 100644 --- a/packages/core/src/progress/progress.spec.ts +++ b/packages/core/src/progress/progress.spec.ts @@ -198,6 +198,34 @@ describe("progress routes", () => { }); }); + it("forwards a complete keyset cursor for subsequent run pages", async () => { + const handler = createProgressHandler() as any; + const event = createEvent( + "/?limit=200&active=true&beforeStartedAt=2026-07-30T12%3A00%3A00.000Z&beforeId=run-0200", + ); + + await handler(event); + + expect(mockListRuns).toHaveBeenCalledWith("boni@local", { + activeOnly: true, + before: { + startedAt: "2026-07-30T12:00:00.000Z", + id: "run-0200", + }, + event, + limit: 200, + }); + }); + + it("rejects incomplete keyset cursors instead of repeating the first page", async () => { + const handler = createProgressHandler() as any; + + await expect( + handler(createEvent("/?beforeStartedAt=2026-07-30T12%3A00%3A00.000Z")), + ).rejects.toMatchObject({ statusCode: 400 }); + expect(mockListRuns).not.toHaveBeenCalled(); + }); + it("short-circuits OPTIONS before auth", async () => { const handler = createProgressHandler() as any; mockGetSession.mockRejectedValue(new Error("should not authenticate")); diff --git a/packages/core/src/progress/routes.ts b/packages/core/src/progress/routes.ts index 8ac16a0304..2e1716a8a4 100644 --- a/packages/core/src/progress/routes.ts +++ b/packages/core/src/progress/routes.ts @@ -3,7 +3,7 @@ * * Mounted under `/_agent-native/runs/*` by `core-routes-plugin`. * - * GET /_agent-native/runs?active=true&limit=50 + * GET /_agent-native/runs?active=true&limit=50&beforeStartedAt=&beforeId= * GET /_agent-native/runs/:id * DELETE /_agent-native/runs/:id * @@ -30,6 +30,22 @@ function parseLimit(value: unknown, fallback = 50): number { return Math.min(Math.floor(n), 200); } +function parseBefore( + startedAt: unknown, + id: unknown, +): { startedAt: string; id: string } | undefined { + if (startedAt == null && id == null) return undefined; + if ( + typeof startedAt !== "string" || + !Number.isFinite(Date.parse(startedAt)) || + typeof id !== "string" || + id.length === 0 + ) { + throw new Error("Invalid progress run cursor"); + } + return { startedAt, id }; +} + async function resolveOwner(event: H3Event): Promise { const session = await getSession(event).catch(() => null); if (!session?.email) { @@ -56,8 +72,20 @@ export function createProgressHandler() { // GET / — list if (method === "GET" && parts.length === 0) { const q = getQuery(event); + let before: ReturnType; + try { + before = parseBefore(q.beforeStartedAt, q.beforeId); + } catch { + const { createError } = await import("h3"); + throw createError({ + statusCode: 400, + statusMessage: + "beforeStartedAt and beforeId must form a valid run cursor", + }); + } return listRuns(owner, { activeOnly: q.active === "true" || q.active === "1", + ...(before ? { before } : {}), limit: parseLimit(q.limit), event, }); diff --git a/packages/core/src/progress/store.spec.ts b/packages/core/src/progress/store.spec.ts index fda77c3256..c5ca5f2fe8 100644 --- a/packages/core/src/progress/store.spec.ts +++ b/packages/core/src/progress/store.spec.ts @@ -66,6 +66,38 @@ describe("progress store", () => { expect(call.args).toEqual(["alice@example.com", 50]); }); + it("uses a deterministic keyset cursor to retrieve rows after the first 200", async () => { + await listRuns("alice@example.com", { + activeOnly: true, + limit: 200, + before: { + startedAt: "2026-07-30T12:00:00.000Z", + id: "run-0200", + }, + }); + + const call = lastSelect(); + expect(call.sql).toMatch( + /started_at < \? OR \(started_at = \? AND id < \?\)/, + ); + expect(call.sql).toMatch(/ORDER BY started_at DESC, id DESC LIMIT \?/); + expect(call.args).toEqual([ + "alice@example.com", + Date.parse("2026-07-30T12:00:00.000Z"), + Date.parse("2026-07-30T12:00:00.000Z"), + "run-0200", + 200, + ]); + }); + + it("rejects an invalid keyset cursor instead of repeating the first page", async () => { + await expect( + listRuns("alice@example.com", { + before: { startedAt: "not-a-timestamp", id: "run-0200" }, + }), + ).rejects.toThrow(/before\.startedAt/); + }); + it("marks stale running rows cancelled before listing active runs", async () => { const now = Date.UTC(2026, 4, 8, 16, 0, 0); vi.spyOn(Date, "now").mockReturnValue(now); diff --git a/packages/core/src/progress/store.ts b/packages/core/src/progress/store.ts index 72d6a7c931..68ee5ed1c0 100644 --- a/packages/core/src/progress/store.ts +++ b/packages/core/src/progress/store.ts @@ -63,8 +63,8 @@ async function ensureTable(): Promise { // avoid ACCESS EXCLUSIVE lock contention in fresh background-worker processes. await ensureTableExists("progress_runs", createSql); await ensureIndexExists( - "idx_progress_runs_owner_status", - `CREATE INDEX IF NOT EXISTS idx_progress_runs_owner_status ON progress_runs (owner, status, started_at)`, + "idx_progress_runs_owner_status_started_id", + `CREATE INDEX IF NOT EXISTS idx_progress_runs_owner_status_started_id ON progress_runs (owner, status, started_at DESC, id DESC)`, ); return; } @@ -78,7 +78,7 @@ async function ensureTable(): Promise { await client.execute(createSql); try { await client.execute( - `CREATE INDEX IF NOT EXISTS idx_progress_runs_owner_status ON progress_runs (owner, status, started_at)`, + `CREATE INDEX IF NOT EXISTS idx_progress_runs_owner_status_started_id ON progress_runs (owner, status, started_at DESC, id DESC)`, ); } catch { // Index already exists or the dialect rejected a duplicate. @@ -331,9 +331,20 @@ export async function listRuns( let where = `owner = ?`; const args: Array = [owner]; if (options.activeOnly) where += ` AND status = 'running'`; + if (options.before) { + const beforeStartedAt = Date.parse(options.before.startedAt); + if (!Number.isFinite(beforeStartedAt)) { + throw new TypeError("before.startedAt must be a valid timestamp"); + } + if (!options.before.id) { + throw new TypeError("before.id must be a non-empty string"); + } + where += ` AND (started_at < ? OR (started_at = ? AND id < ?))`; + args.push(beforeStartedAt, beforeStartedAt, options.before.id); + } args.push(limit); const { rows } = await client.execute({ - sql: `SELECT * FROM progress_runs WHERE ${where} ORDER BY started_at DESC LIMIT ?`, + sql: `SELECT * FROM progress_runs WHERE ${where} ORDER BY started_at DESC, id DESC LIMIT ?`, args, }); return rows.map((r) => parseRow(r as Record)); diff --git a/packages/core/src/progress/types.ts b/packages/core/src/progress/types.ts index 8ddc08da8a..47882658bf 100644 --- a/packages/core/src/progress/types.ts +++ b/packages/core/src/progress/types.ts @@ -54,6 +54,11 @@ export interface ListRunsOptions { activeOnly?: boolean; /** Max rows. Default 50. */ limit?: number; + /** + * Return rows strictly older than this run in `(startedAt, id)` order. + * Pass the final row from the previous page to continue listing. + */ + before?: Pick; /** Optional request event for producers that need to self-dispatch work. */ event?: unknown; }