fabro(01KQT9NFG90GWYZ7CZ0FAH0E12): simplify_opus (succeeded)

Fabro-Run: 01KQT9NFG90GWYZ7CZ0FAH0E12
Fabro-Completed: 6
Fabro-Checkpoint: 6b126321f358fe0eeba6dbfdabe6e41821301cc7

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

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

@@ -218,10 +218,7 @@ export async function apiPaginatedFetcher<TItem, TExtra extends object = {}>(
 }
 
 function stageEventsPagePath(key: string, sinceSeq: number, limit: number): string {
-  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}`;
+  return `${apiPath(key)}?since_seq=${sinceSeq}&limit=${limit}`;
 }
 
 /**

apps/fabro-web/app/lib/query-keys.ts

@@ -44,11 +44,8 @@ export const queryKeys = {
       }),
     events: (id: string, limit = 1000) =>
       withQuery(`/api/v1/runs/${pathSegment(id)}/events`, { limit }),
-    stageEvents: (id: string, stageId: string, sinceSeq?: number, limit?: number) =>
-      withQuery(
-        `/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/events`,
-        { since_seq: sinceSeq, limit },
-      ),
+    stageEvents: (id: string, stageId: string) =>
+      `/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/events`,
     stageLog: (
       id: string,
       stageId: string,

apps/fabro-web/app/lib/run-events.ts

@@ -43,19 +43,25 @@ const RUN_SUMMARY_EVENTS = new Set([
   "run.unarchived",
 ]);
 const STAGE_EVENTS = new Set(["stage.started", "stage.completed", "stage.failed"]);
-// Every event type the `eventsToActivity` reducer in `routes/run-stages.tsx`
-// consumes. When any of these arrive for a stage we currently view, the
-// stage-events SWR key for that stage must be invalidated so the panel
-// refetches. The lifecycle `STAGE_EVENTS` set is kept separate because it
-// also fans out to run-scoped invalidations (stages list, graph, detail).
-const STAGE_ACTIVITY_EVENTS = new Set([
+// Single source of truth: every event type the `eventsToActivity` reducer in
+// `routes/run-stages.tsx` consumes. When any of these arrive for a stage we
+// currently view, the stage-events SWR key for that stage must be invalidated
+// so the panel refetches. The reducer imports this list so the switch stays
+// in sync with the invalidation set; if the reducer grows a new case, this
+// list is the single edit point.
+//
+// The lifecycle `STAGE_EVENTS` set is kept separate because it also fans out
+// to run-scoped invalidations (stages list, graph, detail).
+export const STAGE_ACTIVITY_EVENT_TYPES = [
   "stage.prompt",
   "agent.message",
   "agent.tool.started",
   "agent.tool.completed",
   "command.started",
   "command.completed",
-]);
+] as const;
+export type StageActivityEventType = (typeof STAGE_ACTIVITY_EVENT_TYPES)[number];
+const STAGE_ACTIVITY_EVENTS = new Set<string>(STAGE_ACTIVITY_EVENT_TYPES);
 const INTERVIEW_EVENTS = new Set([
   "interview.started",
   "interview.completed",

apps/fabro-web/app/routes/run-stages.test.ts

@@ -97,18 +97,26 @@ describe("eventsToActivity", () => {
     }
   });
 
-  test("filters out events for other node_ids", () => {
+  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.
     const events: EventEnvelope[] = [
       envelope(1, {
-        event: "agent.message",
-        node_id: "other-stage",
-        properties: { text: "noise" },
+        event: "stage.started",
+        node_id: "detect-drift",
+        properties: {},
       }),
       envelope(2, {
         event: "agent.message",
         node_id: "detect-drift",
         properties: { text: "signal" },
       }),
+      envelope(3, {
+        event: "run.running",
+        node_id: "detect-drift",
+        properties: {},
+      }),
     ];
 
     const turns = eventsToActivity(events, "detect-drift");

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

@@ -41,6 +41,7 @@ import { EmptyState } from "../components/state";
 import { CopyButton } from "../components/ui";
 import { formatDurationSecs } from "../lib/format";
 import { fetchRunCommandLog, useRunStageEvents, useRunStages } from "../lib/queries";
+import { STAGE_ACTIVITY_EVENT_TYPES, type StageActivityEventType } from "../lib/run-events";
 import { mapRunStagesToSidebarStages } from "../lib/stage-sidebar";
 import { getNumber, getString, type UnknownRecord } from "../lib/unknown";
 import {
@@ -65,17 +66,23 @@ function readTermination(props: UnknownRecord): CommandTermination {
   return CommandTermination.EXITED;
 }
 
+const STAGE_ACTIVITY_EVENT_SET = new Set<string>(STAGE_ACTIVITY_EVENT_TYPES);
+
 export function eventsToActivity(events: EventEnvelope[], stageId: string): TurnType[] {
-  const stageEvents = events.filter((e) => e.node_id === stageId);
   const turns: TurnType[] = [];
   // Collect tool pairs: started → completed
   const pendingTools = new Map<string, { toolName: string; input: string }>();
   // Track pending command for pairing started → completed
   let pendingCommand: { stageId: string; script: string; language: string } | undefined;
 
-  for (const e of stageEvents) {
+  for (const e of events) {
+    if (!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.
+    const eventType = e.event as StageActivityEventType;
     const props = e.properties ?? {};
-    switch (e.event) {
+    switch (eventType) {
       case "stage.prompt":
         turns.push({ kind: "system", content: getString(props, "text") ?? e.text ?? "" });
         break;
@@ -567,13 +574,14 @@ export default function RunStages() {
   );
 
   const selectedStage = stages.find((s: Stage) => s.id === stageId) ?? stages[0];
-  const stageEventsQuery = useRunStageEvents(id, selectedStage?.id);
+  const selectedStageId = selectedStage?.id;
+  const stageEventsQuery = useRunStageEvents(id, selectedStageId);
   const turns = useMemo(
     () =>
-      selectedStage
-        ? eventsToActivity(stageEventsQuery.data ?? [], selectedStage.id)
+      selectedStageId
+        ? eventsToActivity(stageEventsQuery.data ?? [], selectedStageId)
         : [],
-    [stageEventsQuery.data, selectedStage],
+    [stageEventsQuery.data, selectedStageId],
   );
   const isRunning = selectedStage?.status === "running";
 

lib/crates/fabro-server/src/demo/mod.rs

@@ -24,7 +24,7 @@ use serde_json::json;
 use crate::error::ApiError;
 use crate::principal_middleware::RequiredUser;
 use crate::run_selector::{ResolveRunError, resolve_run_by_selector};
-use crate::server::{AppState, PaginationParams};
+use crate::server::{AppState, EventListParams, PaginationParams};
 
 fn paginated_response<T: serde::Serialize>(
     items: Vec<T>,
@@ -134,29 +134,11 @@ pub(crate) async fn get_run_stages(
     paginated_response(runs::stages(), &pagination)
 }
 
-#[derive(serde::Deserialize)]
-pub(crate) struct DemoEventListParams {
-    #[serde(default)]
-    since_seq: Option<u32>,
-    #[serde(default)]
-    limit:     Option<usize>,
-}
-
-impl DemoEventListParams {
-    fn since_seq(&self) -> u32 {
-        self.since_seq.unwrap_or(1).max(1)
-    }
-
-    fn limit(&self) -> usize {
-        self.limit.unwrap_or(100).clamp(1, 1000)
-    }
-}
-
 pub(crate) async fn get_stage_events(
     _auth: RequiredUser,
     State(_state): State<Arc<AppState>>,
     Path((_id, stage_id)): Path<(String, String)>,
-    Query(params): Query<DemoEventListParams>,
+    Query(params): Query<EventListParams>,
 ) -> Response {
     let since_seq = params.since_seq();
     let limit = params.limit();

lib/crates/fabro-server/src/server.rs

@@ -138,6 +138,7 @@ use crate::{
 
 mod handler;
 
+pub(crate) use handler::events::EventListParams;
 #[cfg(test)]
 pub(in crate::server) use handler::events::filtered_global_events;
 pub(crate) use handler::graph::render_graph_bytes;

lib/crates/fabro-server/src/server/handler/events.rs

@@ -23,7 +23,7 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
 }
 
 #[derive(serde::Deserialize)]
-struct EventListParams {
+pub(crate) struct EventListParams {
     #[serde(default)]
     since_seq: Option<u32>,
     #[serde(default)]
@@ -31,11 +31,11 @@ struct EventListParams {
 }
 
 impl EventListParams {
-    fn since_seq(&self) -> u32 {
+    pub(crate) fn since_seq(&self) -> u32 {
         self.since_seq.unwrap_or(1).max(1)
     }
 
-    fn limit(&self) -> usize {
+    pub(crate) fn limit(&self) -> usize {
         self.limit.unwrap_or(100).clamp(1, 1000)
     }
 }

lib/crates/fabro-store/src/slate/run_store.rs

@@ -385,6 +385,17 @@ where
     // Unbounded scan first: filtering by node_id with a generic
     // limit-bounded scan would silently drop matches whenever the stage's
     // events are sparse late in the event log.
+    //
+    // We probe just the `node_id` field with a small partial deserialize and
+    // only run the full `RunEvent` parse on matches. Most events in a run
+    // belong to other nodes, so this avoids deserializing large payloads
+    // (`agent.tool.completed.output`, `agent.message.text`, …) we'd discard.
+    #[derive(serde::Deserialize)]
+    struct NodeIdProbe<'a> {
+        #[serde(default, borrow)]
+        node_id: Option<&'a str>,
+    }
+
     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? {
@@ -395,10 +406,11 @@ where
         if seq < start_seq {
             continue;
         }
-        let event: RunEvent = serde_json::from_slice(&entry.value)?;
-        if event.node_id.as_deref() != Some(node_id) {
+        let probe: NodeIdProbe = serde_json::from_slice(&entry.value)?;
+        if probe.node_id != Some(node_id) {
             continue;
         }
+        let event: RunEvent = serde_json::from_slice(&entry.value)?;
         events.push(EventEnvelope { seq, event });
     }
     events.sort_by_key(|event| event.seq);