From 676dd52800b773aaa94db183ec5561b3ac4c0c94 Mon Sep 17 00:00:00 2001 From: George Pickett Date: Sun, 1 Mar 2026 23:35:17 -0800 Subject: [PATCH] fix: keep history cursor stable while backfill is incomplete --- .../runtime/agents/[agentId]/history/route.ts | 67 ++++++++++++++++--- tests/unit/runtimeRoutes.test.ts | 53 ++++++++++++++- 2 files changed, 107 insertions(+), 13 deletions(-) diff --git a/src/app/api/runtime/agents/[agentId]/history/route.ts b/src/app/api/runtime/agents/[agentId]/history/route.ts index c75237e..d3f628f 100644 --- a/src/app/api/runtime/agents/[agentId]/history/route.ts +++ b/src/app/api/runtime/agents/[agentId]/history/route.ts @@ -1,13 +1,14 @@ import { NextResponse } from "next/server"; import { deriveRuntimeFreshness, probeOpenClawLocalState } from "@/lib/controlplane/degraded-read"; -import { selectAgentHistoryEntries } from "@/lib/controlplane/read-model"; -import { getControlPlaneRuntime, isStudioDomainApiModeEnabled } from "@/lib/controlplane/runtime"; +import { bootstrapDomainRuntime } from "@/lib/controlplane/runtime-route-bootstrap"; export const runtime = "nodejs"; const DEFAULT_LIMIT = 200; const MAX_LIMIT = 1000; +const BACKFILL_BATCH_LIMIT = 500; +const MAX_BACKFILL_BATCHES_PER_REQUEST = 2; const resolveLimit = (raw: string | null): number => { if (!raw) return DEFAULT_LIMIT; @@ -17,11 +18,20 @@ const resolveLimit = (raw: string | null): number => { return Math.min(Math.floor(parsed), MAX_LIMIT); }; +const resolveBeforeOutboxId = (raw: string | null, fallback: number): number => { + if (!raw) return fallback; + const parsed = Number(raw); + if (!Number.isFinite(parsed)) return fallback; + if (parsed <= 0) return fallback; + return Math.min(Math.floor(parsed), fallback); +}; + export async function GET( request: Request, context: { params: Promise<{ agentId: string }> } ) { - if (!isStudioDomainApiModeEnabled()) { + const bootstrap = await bootstrapDomainRuntime(); + if (bootstrap.kind === "mode-disabled") { return NextResponse.json({ enabled: false, error: "domain_api_mode_disabled" }, { status: 404 }); } @@ -31,26 +41,61 @@ export async function GET( return NextResponse.json({ error: "agentId is required." }, { status: 400 }); } - const controlPlane = getControlPlaneRuntime(); - let startError: string | null = null; - try { - await controlPlane.ensureStarted(); - } catch (err) { - startError = err instanceof Error ? err.message : "controlplane_start_failed"; + if (bootstrap.kind === "runtime-init-failed") { + return NextResponse.json( + { + enabled: true, + error: bootstrap.message, + code: "CONTROLPLANE_RUNTIME_INIT_FAILED", + reason: "runtime_init_failed", + }, + { status: 503 } + ); } + const controlPlane = bootstrap.runtime; + const startError = bootstrap.kind === "start-failed" ? bootstrap.message : null; const url = new URL(request.url); const limit = resolveLimit(url.searchParams.get("limit")); const snapshot = controlPlane.snapshot(); + const beforeOutboxId = resolveBeforeOutboxId( + url.searchParams.get("beforeOutboxId"), + snapshot.outboxHead + 1 + ); const probe = snapshot.status === "connected" ? null : await probeOpenClawLocalState(); - const allEntries = controlPlane.eventsAfter(0, MAX_LIMIT * 5); - const entries = selectAgentHistoryEntries(allEntries, normalizedAgentId, limit); + let results = controlPlane.eventsBeforeForAgent(normalizedAgentId, beforeOutboxId, limit + 1); + let backfillIncomplete = false; + if (results.length <= limit) { + for (let attempt = 0; attempt < MAX_BACKFILL_BATCHES_PER_REQUEST; attempt += 1) { + const backfill = controlPlane.backfillAgentHistoryIndex(beforeOutboxId, BACKFILL_BATCH_LIMIT); + if (backfill.scannedRows === 0) { + backfillIncomplete = false; + break; + } + backfillIncomplete = !backfill.exhausted; + results = controlPlane.eventsBeforeForAgent(normalizedAgentId, beforeOutboxId, limit + 1); + if (results.length > limit || backfill.exhausted) { + break; + } + } + } + const hasMore = results.length > limit || backfillIncomplete; + const entries = results.length > limit ? results.slice(results.length - limit) : results; + const nextBeforeOutboxId = hasMore + ? backfillIncomplete + ? beforeOutboxId + : entries.length > 0 + ? entries[0].id + : null + : null; return NextResponse.json({ enabled: true, agentId: normalizedAgentId, ...(startError ? { error: startError } : {}), entries, + hasMore, + nextBeforeOutboxId, freshness: deriveRuntimeFreshness(snapshot, probe), ...(probe ? { probe } : {}), }); diff --git a/tests/unit/runtimeRoutes.test.ts b/tests/unit/runtimeRoutes.test.ts index ba34950..cea644f 100644 --- a/tests/unit/runtimeRoutes.test.ts +++ b/tests/unit/runtimeRoutes.test.ts @@ -287,7 +287,7 @@ describe("runtime routes", () => { expect(secondBody.nextBeforeOutboxId).toBeNull(); }); - it("agent history route backfills legacy rows in bounded batches", async () => { + it("agent history route keeps cursor stable while backfill remains incomplete", async () => { const eventsBeforeForAgent = vi .fn() .mockReturnValueOnce([ @@ -380,7 +380,56 @@ describe("runtime routes", () => { expect(eventsBeforeForAgent).toHaveBeenNthCalledWith(2, "alpha", 11, 3); expect(body.entries.map((entry) => entry.id)).toEqual([5, 7]); expect(body.hasMore).toBe(true); - expect(body.nextBeforeOutboxId).toBe(5); + expect(body.nextBeforeOutboxId).toBe(11); + }); + + it("agent history route reports continuation when backfill cap is reached before exhaustion", async () => { + const eventsBeforeForAgent = vi.fn().mockReturnValue([]); + const backfillAgentHistoryIndex = vi + .fn() + .mockReturnValue({ scannedRows: 500, updatedRows: 0, exhausted: false }); + const runtimeMock: RuntimeMock = { + ensureStarted: async () => {}, + snapshot: () => ({ + status: "connected", + reason: null, + asOf: "2026-02-28T02:40:00.000Z", + outboxHead: 10, + }), + eventsAfter: () => [], + eventsBefore: () => { + throw new Error("legacy scan path used"); + }, + eventsBeforeForAgent, + backfillAgentHistoryIndex, + subscribe: () => () => {}, + }; + + const mod = await loadRouteModule<{ + GET: ( + request: Request, + context: { params: Promise<{ agentId: string }> } + ) => Promise; + }>("@/app/api/runtime/agents/[agentId]/history/route", runtimeMock); + + const response = await mod.GET( + new Request("http://localhost/api/runtime/agents/alpha/history?limit=2"), + { params: Promise.resolve({ agentId: "alpha" }) } + ); + expect(response.status).toBe(200); + const body = (await response.json()) as { + entries: Array<{ id: number }>; + hasMore: boolean; + nextBeforeOutboxId: number | null; + }; + expect(eventsBeforeForAgent).toHaveBeenNthCalledWith(1, "alpha", 11, 3); + expect(eventsBeforeForAgent).toHaveBeenNthCalledWith(2, "alpha", 11, 3); + expect(eventsBeforeForAgent).toHaveBeenNthCalledWith(3, "alpha", 11, 3); + expect(backfillAgentHistoryIndex).toHaveBeenNthCalledWith(1, 11, 500); + expect(backfillAgentHistoryIndex).toHaveBeenNthCalledWith(2, 11, 500); + expect(body.entries).toEqual([]); + expect(body.hasMore).toBe(true); + expect(body.nextBeforeOutboxId).toBe(11); }); it("stream route replays from Last-Event-ID and emits live updates", async () => {