fabro(01KQT9NFG90GWYZ7CZ0FAH0E12): simplify_gpt (succeeded)

Fabro-Run: 01KQT9NFG90GWYZ7CZ0FAH0E12
Fabro-Completed: 7

⚒️ Generated with [Fabro](https://fabro.sh)

apps/fabro-web/app/lib/api-client.ts

@@ -218,7 +218,10 @@ export async function apiPaginatedFetcher<TItem, TExtra extends object = {}>(
 }
 
 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<TItem>(key: string): Promise<TItem[]> {
+export async function fetchAllStageEvents<TItem extends { seq: number }>(
+  key: string,
+): Promise<TItem[]> {
   const PAGE_LIMIT = 1000;
   const MAX_PAGES = 50;
   const data: TItem[] = [];
@@ -242,7 +247,7 @@ export async function fetchAllStageEvents<TItem>(key: string): Promise<TItem[]>
     if (!response.ok) {
       throw await apiErrorFromResponse(response);
     }
-    const page = (await response.json()) as PaginatedEnvelope<TItem & { seq: number }>;
+    const page = (await response.json()) as PaginatedEnvelope<TItem>;
     pagesLoaded += 1;
 
     if (page.data.length === 0) {
@@ -264,8 +269,14 @@ export async function fetchAllStageEvents<TItem>(key: string): Promise<TItem[]>
       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;
   }
 }
 

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", () => {

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,

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");

apps/fabro-web/app/routes/run-stages.tsx

@@ -68,6 +68,10 @@ function readTermination(props: UnknownRecord): CommandTermination {
 
 const STAGE_ACTIVITY_EVENT_SET = new Set<string>(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);
     }
   }
 

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<EventEnvelope> = 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)
 }