From 4c5fba30331b95f1d8d3f337b3ced3c377877f0f Mon Sep 17 00:00:00 2001 From: Fabro Date: Mon, 4 May 2026 20:53:16 +0000 Subject: [PATCH] fabro(01KQT9NFG90GWYZ7CZ0FAH0E12): simplify_opus (succeeded) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fabro-Run: 01KQT9NFG90GWYZ7CZ0FAH0E12 Fabro-Completed: 6 Fabro-Checkpoint: 6b126321f358fe0eeba6dbfdabe6e41821301cc7 ⚒️ Generated with [Fabro](https://fabro.sh) --- apps/fabro-web/app/lib/api-client.ts | 5 +---- apps/fabro-web/app/lib/query-keys.ts | 7 ++---- apps/fabro-web/app/lib/run-events.ts | 20 +++++++++++------ apps/fabro-web/app/routes/run-stages.test.ts | 16 ++++++++++---- apps/fabro-web/app/routes/run-stages.tsx | 22 +++++++++++++------ lib/crates/fabro-server/src/demo/mod.rs | 22 ++----------------- lib/crates/fabro-server/src/server.rs | 1 + .../fabro-server/src/server/handler/events.rs | 6 ++--- lib/crates/fabro-store/src/slate/run_store.rs | 16 ++++++++++++-- 9 files changed, 63 insertions(+), 52 deletions(-) diff --git a/apps/fabro-web/app/lib/api-client.ts b/apps/fabro-web/app/lib/api-client.ts index 5c57896a7..257bc49ab 100644 --- a/apps/fabro-web/app/lib/api-client.ts +++ b/apps/fabro-web/app/lib/api-client.ts @@ -218,10 +218,7 @@ export async function apiPaginatedFetcher( } 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}`; } /** diff --git a/apps/fabro-web/app/lib/query-keys.ts b/apps/fabro-web/app/lib/query-keys.ts index 808cbcd56..34dee2c9b 100644 --- a/apps/fabro-web/app/lib/query-keys.ts +++ b/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, diff --git a/apps/fabro-web/app/lib/run-events.ts b/apps/fabro-web/app/lib/run-events.ts index 8661f5142..7ab282100 100644 --- a/apps/fabro-web/app/lib/run-events.ts +++ b/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(STAGE_ACTIVITY_EVENT_TYPES); const INTERVIEW_EVENTS = new Set([ "interview.started", "interview.completed", diff --git a/apps/fabro-web/app/routes/run-stages.test.ts b/apps/fabro-web/app/routes/run-stages.test.ts index 4d5a76ce2..947f68bdd 100644 --- a/apps/fabro-web/app/routes/run-stages.test.ts +++ b/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"); diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index d65728f1d..30bad11b4 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/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(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(); // 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"; diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index 697d76865..93d1546f3 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/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( items: Vec, @@ -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, - #[serde(default)] - limit: Option, -} - -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>, Path((_id, stage_id)): Path<(String, String)>, - Query(params): Query, + Query(params): Query, ) -> Response { let since_seq = params.since_seq(); let limit = params.limit(); diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 9c75acb5f..529584fae 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/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; diff --git a/lib/crates/fabro-server/src/server/handler/events.rs b/lib/crates/fabro-server/src/server/handler/events.rs index a25888345..d9f6a23ff 100644 --- a/lib/crates/fabro-server/src/server/handler/events.rs +++ b/lib/crates/fabro-server/src/server/handler/events.rs @@ -23,7 +23,7 @@ pub(super) fn routes() -> Router> { } #[derive(serde::Deserialize)] -struct EventListParams { +pub(crate) struct EventListParams { #[serde(default)] since_seq: Option, #[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) } } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 3bc054bc7..3bb48d8bb 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/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 = 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);