diff --git a/apps/fabro-web/app/lib/api-client.ts b/apps/fabro-web/app/lib/api-client.ts index 257bc49ab..87792ed69 100644 --- a/apps/fabro-web/app/lib/api-client.ts +++ b/apps/fabro-web/app/lib/api-client.ts @@ -218,7 +218,10 @@ export async function apiPaginatedFetcher( } function stageEventsPagePath(key: string, sinceSeq: number, limit: number): string { - return `${apiPath(key)}?since_seq=${sinceSeq}&limit=${limit}`; + const url = new URL(apiPath(key), "http://fabro.local"); + url.searchParams.set("since_seq", String(sinceSeq)); + url.searchParams.set("limit", String(limit)); + return `${url.pathname}${url.search}`; } /** @@ -230,7 +233,9 @@ function stageEventsPagePath(key: string, sinceSeq: number, limit: number): stri * `has_more` but returns no rows we exit and `console.warn` to surface the * server invariant violation without spinning the UI. */ -export async function fetchAllStageEvents(key: string): Promise { +export async function fetchAllStageEvents( + key: string, +): Promise { const PAGE_LIMIT = 1000; const MAX_PAGES = 50; const data: TItem[] = []; @@ -242,7 +247,7 @@ export async function fetchAllStageEvents(key: string): Promise if (!response.ok) { throw await apiErrorFromResponse(response); } - const page = (await response.json()) as PaginatedEnvelope; + const page = (await response.json()) as PaginatedEnvelope; pagesLoaded += 1; if (page.data.length === 0) { @@ -264,8 +269,14 @@ export async function fetchAllStageEvents(key: string): Promise return data; } - const lastSeq = page.data[page.data.length - 1].seq; - sinceSeq = lastSeq + 1; + const highestSeq = page.data.reduce((max, event) => Math.max(max, event.seq), sinceSeq - 1); + if (highestSeq < sinceSeq) { + console.warn( + `Stage events fetch for ${key} returned a non-advancing page at since_seq=${sinceSeq}; stopping at ${data.length} items to avoid spinning.`, + ); + return data; + } + sinceSeq = highestSeq + 1; } } diff --git a/apps/fabro-web/app/lib/query-keys.test.ts b/apps/fabro-web/app/lib/query-keys.test.ts index b82695872..5d7b3a53a 100644 --- a/apps/fabro-web/app/lib/query-keys.test.ts +++ b/apps/fabro-web/app/lib/query-keys.test.ts @@ -11,6 +11,9 @@ describe("queryKeys", () => { expect(queryKeys.runs.stageLog("run 1", "build step@2", "stderr", 12, 34)).toBe( "/api/v1/runs/run%201/stages/build%20step%402/logs/stderr?offset=12&limit=34", ); + expect(queryKeys.runs.stageEvents("run 1", "build step", 7, 25)).toBe( + "/api/v1/runs/run%201/stages/build%20step/events?since_seq=7&limit=25", + ); }); test("event-mapped keys match query hook resources", () => { diff --git a/apps/fabro-web/app/lib/query-keys.ts b/apps/fabro-web/app/lib/query-keys.ts index 34dee2c9b..402739897 100644 --- a/apps/fabro-web/app/lib/query-keys.ts +++ b/apps/fabro-web/app/lib/query-keys.ts @@ -44,8 +44,11 @@ export const queryKeys = { }), events: (id: string, limit = 1000) => withQuery(`/api/v1/runs/${pathSegment(id)}/events`, { limit }), - stageEvents: (id: string, stageId: string) => - `/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/events`, + stageEvents: (id: string, stageId: string, sinceSeq?: number, limit?: number) => + withQuery(`/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/events`, { + since_seq: sinceSeq, + limit, + }), stageLog: ( id: string, stageId: string, diff --git a/apps/fabro-web/app/routes/run-stages.test.ts b/apps/fabro-web/app/routes/run-stages.test.ts index 947f68bdd..856ce325d 100644 --- a/apps/fabro-web/app/routes/run-stages.test.ts +++ b/apps/fabro-web/app/routes/run-stages.test.ts @@ -97,10 +97,9 @@ describe("eventsToActivity", () => { } }); - test("ignores events of unknown types", () => { - // The reducer only consumes STAGE_ACTIVITY_EVENT_TYPES; lifecycle and - // unrelated events are skipped. The server scopes the input to a single - // node, so node_id filtering is the server's responsibility. + test("ignores unknown event types and events for other nodes", () => { + // The reducer defensively filters by node_id even though the server scopes + // this endpoint, and only consumes STAGE_ACTIVITY_EVENT_TYPES. const events: EventEnvelope[] = [ envelope(1, { event: "stage.started", @@ -117,6 +116,11 @@ describe("eventsToActivity", () => { node_id: "detect-drift", properties: {}, }), + envelope(4, { + event: "agent.message", + node_id: "other-stage", + properties: { text: "wrong stage" }, + }), ]; const turns = eventsToActivity(events, "detect-drift"); diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index 30bad11b4..78fa6a9ef 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/apps/fabro-web/app/routes/run-stages.tsx @@ -68,6 +68,10 @@ function readTermination(props: UnknownRecord): CommandTermination { const STAGE_ACTIVITY_EVENT_SET = new Set(STAGE_ACTIVITY_EVENT_TYPES); +function assertNever(value: never): never { + throw new Error(`Unhandled stage activity event type: ${value}`); +} + export function eventsToActivity(events: EventEnvelope[], stageId: string): TurnType[] { const turns: TurnType[] = []; // Collect tool pairs: started → completed @@ -76,7 +80,7 @@ export function eventsToActivity(events: EventEnvelope[], stageId: string): Turn let pendingCommand: { stageId: string; script: string; language: string } | undefined; for (const e of events) { - if (!STAGE_ACTIVITY_EVENT_SET.has(e.event)) continue; + if (e.node_id !== stageId || !STAGE_ACTIVITY_EVENT_SET.has(e.event)) continue; // Exhaustive switch over StageActivityEventType: adding a new variant to // STAGE_ACTIVITY_EVENT_TYPES forces a TS error here until the case is // handled, keeping the SWR invalidation set and the reducer in sync. @@ -140,6 +144,8 @@ export function eventsToActivity(events: EventEnvelope[], stageId: string): Turn pendingCommand = undefined; break; } + default: + assertNever(eventType); } } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 3bb48d8bb..ed68442b5 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -396,6 +396,7 @@ where node_id: Option<&'a str>, } + let max_events = limit.saturating_add(1); let mut iter = db.scan_prefix(keys::run_events_prefix(run_id)).await?; let mut events: Vec = Vec::new(); while let Some(entry) = iter.next().await? { @@ -411,10 +412,24 @@ where continue; } let event: RunEvent = serde_json::from_slice(&entry.value)?; - events.push(EventEnvelope { seq, event }); + let envelope = EventEnvelope { seq, event }; + if events.len() < max_events { + events.push(envelope); + continue; + } + + if let Some((max_index, max_seq)) = events + .iter() + .enumerate() + .max_by_key(|(_, existing)| existing.seq) + .map(|(index, existing)| (index, existing.seq)) + { + if seq < max_seq { + events[max_index] = envelope; + } + } } events.sort_by_key(|event| event.seq); - events.truncate(limit.saturating_add(1)); Ok(events) }