From 6e36d8350ed859949f8d0c39d97ef31f04f246c3 Mon Sep 17 00:00:00 2001 From: "fabro-sh-0530[bot]" <281434857+fabro-sh-0530[bot]@users.noreply.github.com> Date: Tue, 5 May 2026 09:16:52 -0400 Subject: [PATCH 1/2] Render stage activity from scoped events endpoint (#212) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Stage detail now loads activity from a canonical stage-scoped events endpoint instead of falling back to the first 1000 run-wide events. This fixes empty panes for late stages in long runs and removes the presentation-shaped `StageTurn` API from the wire. ### Plan Summary - Add `GET /runs/{id}/stages/{stageId}/events` with cursor pagination and server-side `node_id` filtering. - Replace frontend stage-turn/fallback loading with paginated stage-events loading and local event-to-activity projection. - Broaden SSE/SWR invalidation so every activity event consumed by the reducer refreshes the per-stage cache. - Remove `StageTurn` schemas/client models and update demo fixtures plus pagination/handler/reducer tests. ## What changed and why The store now scans the run event prefix and filters by `node_id` before applying the `limit + 1` cutoff. That preserves sparse late-stage matches that would otherwise be dropped if we reused the run-wide limited scan and filtered afterward. The real-mode handler returns an empty page for an unknown stage id in an existing run, while preserving 404 for missing runs. On the frontend, `run-stages` fetches all pages for the selected stage and feeds them through `eventsToActivity`, keeping `TurnType` as a local presentation model. Invalidation now targets `runs.stageEvents(runId, stageId)` for lifecycle and reducer-consumed activity events (`stage.prompt`, agent messages/tools, and command events), so active panes refresh from the existing run event subscription. The OpenAPI document and generated TS client now expose `listStageEvents` and drop stale `StageTurn` models. Demo mode serves a `detect-drift` stage-events fixture using the same cursor semantics as the real endpoint. ## API notes `/runs/{id}/stages/{stageId}/turns` is removed; clients should use `/runs/{id}/stages/{stageId}/events?since_seq=&limit=` and project events locally. The `stageId` path segment for this endpoint is the workflow node id, not the visit-qualified `node_id@visit` form used by command logs/artifacts. ⚒️ Generated with [Fabro](https://fabro.sh) --------- Co-authored-by: Fabro Co-authored-by: Bryan Helmkamp --- apps/fabro-web/app/lib/api-client.ts | 65 +++- apps/fabro-web/app/lib/queries.ts | 25 +- apps/fabro-web/app/lib/query-keys.test.ts | 26 +- apps/fabro-web/app/lib/query-keys.ts | 9 +- apps/fabro-web/app/lib/run-events.test.tsx | 10 +- apps/fabro-web/app/lib/run-events.ts | 33 +- apps/fabro-web/app/routes/run-stages.test.ts | 139 ++++++-- apps/fabro-web/app/routes/run-stages.tsx | 88 +++-- docs/public/api-reference/fabro-api.yaml | 129 +------- lib/crates/fabro-server/src/demo/mod.rs | 159 ++++++++- .../fabro-server/src/principal_middleware.rs | 18 + lib/crates/fabro-server/src/server.rs | 3 +- .../fabro-server/src/server/handler/events.rs | 311 +++++++++++++++++- .../fabro-server/src/server/handler/mod.rs | 5 +- .../fabro-server/tests/it/event_pagination.rs | 75 +++++ lib/crates/fabro-server/tests/it/main.rs | 1 + .../fabro-server/tests/it/pagination.rs | 4 - lib/crates/fabro-store/src/slate/run_store.rs | 285 +++++++++++++++- .../src/.openapi-generator/FILES | 6 - .../src/api/run-internals-api.ts | 64 ++-- .../src/models/assistant-stage-turn.ts | 34 -- .../src/models/board-column-definition.ts | 3 + .../fabro-api-client/src/models/index.ts | 6 - .../src/models/manifest-git.ts | 41 --- .../models/manifest-pre-run-push-outcome.ts | 36 -- .../src/models/paginated-stage-turn-list.ts | 30 -- .../src/models/pre-run-git-context.ts | 30 -- .../src/models/set-secret-request.ts | 25 -- .../fabro-api-client/src/models/stage-turn.ts | 35 -- .../src/models/system-stage-turn.ts | 34 -- .../src/models/tool-stage-turn.ts | 41 --- .../fabro-api-client/src/models/tool-use.ts | 46 --- 32 files changed, 1153 insertions(+), 663 deletions(-) create mode 100644 lib/crates/fabro-server/tests/it/event_pagination.rs delete mode 100644 lib/packages/fabro-api-client/src/models/assistant-stage-turn.ts delete mode 100644 lib/packages/fabro-api-client/src/models/manifest-git.ts delete mode 100644 lib/packages/fabro-api-client/src/models/manifest-pre-run-push-outcome.ts delete mode 100644 lib/packages/fabro-api-client/src/models/paginated-stage-turn-list.ts delete mode 100644 lib/packages/fabro-api-client/src/models/pre-run-git-context.ts delete mode 100644 lib/packages/fabro-api-client/src/models/set-secret-request.ts delete mode 100644 lib/packages/fabro-api-client/src/models/stage-turn.ts delete mode 100644 lib/packages/fabro-api-client/src/models/system-stage-turn.ts delete mode 100644 lib/packages/fabro-api-client/src/models/tool-stage-turn.ts delete mode 100644 lib/packages/fabro-api-client/src/models/tool-use.ts diff --git a/apps/fabro-web/app/lib/api-client.ts b/apps/fabro-web/app/lib/api-client.ts index bd990361b..87792ed69 100644 --- a/apps/fabro-web/app/lib/api-client.ts +++ b/apps/fabro-web/app/lib/api-client.ts @@ -217,6 +217,69 @@ 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}`; +} + +/** + * Cursor-paginated fetcher for `/runs/{id}/stages/{stageId}/events`. + * + * Loops from `since_seq=1` with a 1000-event page size, advancing the cursor + * to `highestSeq + 1` until the server reports `meta.has_more === false`. + * The empty-page guard mirrors `apiPaginatedFetcher`: if the server claims + * `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 { + const PAGE_LIMIT = 1000; + const MAX_PAGES = 50; + const data: TItem[] = []; + let sinceSeq = 1; + let pagesLoaded = 0; + + while (true) { + const response = await apiRequest(stageEventsPagePath(key, sinceSeq, PAGE_LIMIT)); + if (!response.ok) { + throw await apiErrorFromResponse(response); + } + const page = (await response.json()) as PaginatedEnvelope; + pagesLoaded += 1; + + if (page.data.length === 0) { + if (page.meta.has_more) { + console.warn( + `Stage events fetch for ${key} returned an empty page with has_more=true; stopping at ${data.length} items to avoid spinning.`, + ); + } + return data; + } + + data.push(...page.data); + if (!page.meta.has_more) return data; + + if (pagesLoaded >= MAX_PAGES) { + console.warn( + `Stopped stage events fetch for ${key} after ${pagesLoaded} pages and ${data.length} items because the safety cap was reached.`, + ); + return data; + } + + 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; + } +} + export async function apiJsonMutation( key: string, { arg }: { arg: TArg }, @@ -233,4 +296,4 @@ export async function apiJsonMutation( } if (response.status === 204) return undefined as TResponse; return response.json() as Promise; -} +} \ No newline at end of file diff --git a/apps/fabro-web/app/lib/queries.ts b/apps/fabro-web/app/lib/queries.ts index 0da17785e..880cf97b6 100644 --- a/apps/fabro-web/app/lib/queries.ts +++ b/apps/fabro-web/app/lib/queries.ts @@ -1,12 +1,11 @@ import useSWR, { type SWRConfiguration } from "swr"; import type { ApiQuestion, + EventEnvelope, PaginatedBoardRunList, - PaginatedEventList, PaginatedRunFileList, PaginatedRunList, PaginatedRunStageList, - PaginatedStageTurnList, CommandLogResponse, CommandOutputStream, RunBilling, @@ -24,6 +23,7 @@ import { apiNullableTextFetcher, apiPaginatedFetcher, apiTextFetcher, + fetchAllStageEvents, type PaginatedEnvelope, } from "./api-client"; import { queryKeys } from "./query-keys"; @@ -140,21 +140,10 @@ export function useRunQuestions(id: string | undefined, enabled: boolean) { ); } -export function useRunStageTurns( - id: string | undefined, - stageId: string | undefined, - enabled = true, -) { - return useSWR( - id && stageId && enabled ? queryKeys.runs.stageTurns(id, stageId) : null, - apiNullableFetcher, - ); -} - -export function useRunEventsList(id: string | undefined, enabled = true) { - return useSWR( - id && enabled ? queryKeys.runs.events(id, 1000) : null, - apiNullableFetcher, +export function useRunStageEvents(id: string | undefined, stageId: string | undefined) { + return useSWR( + id && stageId ? queryKeys.runs.stageEvents(id, stageId) : null, + fetchAllStageEvents, ); } @@ -205,4 +194,4 @@ export function useServerSettings() { return useSWR(queryKeys.settings.server(), apiFetcher, immutableOptions); } -export { apiTextFetcher }; +export { apiTextFetcher }; \ No newline at end of file diff --git a/apps/fabro-web/app/lib/query-keys.test.ts b/apps/fabro-web/app/lib/query-keys.test.ts index 8f036ac3e..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", () => { @@ -23,7 +26,26 @@ describe("queryKeys", () => { queryKeys.runs.graph("run-1", "LR"), queryKeys.runs.graph("run-1", "TB"), queryKeys.runs.detail("run-1"), - queryKeys.runs.stageTurns("run-1", "stage-1"), + queryKeys.runs.stageEvents("run-1", "stage-1"), ]); }); -}); + + test("agent activity events invalidate the per-stage events key", () => { + for (const event of [ + "stage.prompt", + "agent.message", + "agent.tool.started", + "agent.tool.completed", + "command.started", + "command.completed", + ]) { + expect(queryKeysForRunEvent("run-1", event, "stage-1")).toEqual([ + queryKeys.runs.stageEvents("run-1", "stage-1"), + ]); + } + }); + + test("agent activity events without a node_id invalidate nothing", () => { + expect(queryKeysForRunEvent("run-1", "agent.message")).toEqual([]); + }); +}); \ No newline at end of file diff --git a/apps/fabro-web/app/lib/query-keys.ts b/apps/fabro-web/app/lib/query-keys.ts index 7d3e847e7..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 }), - stageTurns: (id: string, stageId: string) => - `/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/turns`, + 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, @@ -75,4 +78,4 @@ export const queryKeys = { settings: { server: () => "/api/v1/settings", }, -}; +}; \ No newline at end of file diff --git a/apps/fabro-web/app/lib/run-events.test.tsx b/apps/fabro-web/app/lib/run-events.test.tsx index c38ea1dd2..14db0e3f2 100644 --- a/apps/fabro-web/app/lib/run-events.test.tsx +++ b/apps/fabro-web/app/lib/run-events.test.tsx @@ -55,7 +55,7 @@ describe("queryKeysForRunEvent", () => { expect(keys).toContain(queryKeys.runs.stages("run-1")); expect(keys).toContain(queryKeys.runs.events("run-1", 1000)); expect(keys).toContain(queryKeys.runs.detail("run-1")); - expect(keys).toContain(queryKeys.runs.stageTurns("run-1", "verify@2")); + expect(keys).toContain(queryKeys.runs.stageEvents("run-1", "verify@2")); }); }); @@ -180,7 +180,7 @@ describe("subscribeToRunEvents", () => { coordinator.close(); }); - test("envelope with suffixed stage_id invalidates stageTurns(runId, stageId)", async () => { + test("envelope with suffixed stage_id invalidates stageEvents(runId, stageId)", async () => { const source = new FakeEventSource(); const keys: string[] = []; const coordinator = createCoordinator(() => source); @@ -202,12 +202,12 @@ describe("subscribeToRunEvents", () => { node_id: "verify", }); - expect(keys).toContain(queryKeys.runs.stageTurns("run-stage", "verify@2")); + expect(keys).toContain(queryKeys.runs.stageEvents("run-stage", "verify@2")); expect(keys).toContain(queryKeys.runs.stages("run-stage")); expect(keys).toContain(queryKeys.runs.events("run-stage", 1000)); expect(keys).toContain(queryKeys.runs.graph("run-stage", "LR")); expect(keys).toContain(queryKeys.runs.detail("run-stage")); - expect(keys).not.toContain(queryKeys.runs.stageTurns("run-stage", "verify")); + expect(keys).not.toContain(queryKeys.runs.stageEvents("run-stage", "verify")); cleanup(); coordinator.close(); @@ -230,7 +230,7 @@ describe("subscribeToRunEvents", () => { await waitFor(() => source.onmessage !== null); source.emit({ event: "stage.started", run_id: "run-stage-node", node_id: "verify" }); - expect(keys).toContain(queryKeys.runs.stageTurns("run-stage-node", "verify")); + expect(keys).toContain(queryKeys.runs.stageEvents("run-stage-node", "verify")); expect(keys).toContain(queryKeys.runs.stages("run-stage-node")); cleanup(); diff --git a/apps/fabro-web/app/lib/run-events.ts b/apps/fabro-web/app/lib/run-events.ts index a81a11c2b..1546d1578 100644 --- a/apps/fabro-web/app/lib/run-events.ts +++ b/apps/fabro-web/app/lib/run-events.ts @@ -49,7 +49,25 @@ const STAGE_EVENTS = new Set([ "stage.failed", "stage.retrying", ]); -const COMMAND_EVENTS = new Set(["command.started", "command.completed"]); +// 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", @@ -97,20 +115,13 @@ export function queryKeysForRunEvent( queryKeys.runs.detail(runId), ]; if (stageId) { - keys.push(queryKeys.runs.stageTurns(runId, stageId)); + keys.push(queryKeys.runs.stageEvents(runId, stageId)); } return keys; } - if (COMMAND_EVENTS.has(event)) { - const keys = [ - queryKeys.runs.stages(runId), - queryKeys.runs.events(runId, 1000), - ]; - if (stageId) { - keys.push(queryKeys.runs.stageTurns(runId, stageId)); - } - return keys; + if (STAGE_ACTIVITY_EVENTS.has(event)) { + return stageId ? [queryKeys.runs.stageEvents(runId, stageId)] : []; } return []; diff --git a/apps/fabro-web/app/routes/run-stages.test.ts b/apps/fabro-web/app/routes/run-stages.test.ts index 704c5cac8..52f10ed05 100644 --- a/apps/fabro-web/app/routes/run-stages.test.ts +++ b/apps/fabro-web/app/routes/run-stages.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; import type { EventEnvelope } from "@qltysh/fabro-api-client"; -import { isSafeMarkdownHref, turnsFromEvents } from "./run-stages"; +import { eventsToActivity, isSafeMarkdownHref } from "./run-stages"; describe("isSafeMarkdownHref", () => { test("rejects protocol-relative URLs", () => { @@ -17,43 +17,39 @@ describe("isSafeMarkdownHref", () => { }); }); -function makeEnvelope(overrides: Partial): EventEnvelope { +function envelope(seq: number, partial: Partial): EventEnvelope { return { - seq: 1, - id: "evt", - ts: "2026-01-01T00:00:00Z", + seq, + id: `evt-${seq}`, + ts: "2026-04-09T12:00:00Z", run_id: "run-1", event: "stage.prompt", - ...overrides, + ...partial, } as EventEnvelope; } -describe("turnsFromEvents", () => { +describe("eventsToActivity", () => { test("filters events by stage_id (verify@1 vs verify@2 do not cross-contaminate)", () => { const events: EventEnvelope[] = [ - makeEnvelope({ - seq: 1, + envelope(1, { event: "stage.prompt", stage_id: "verify@1", node_id: "verify", properties: { text: "first visit prompt" }, }), - makeEnvelope({ - seq: 2, + envelope(2, { event: "stage.prompt", stage_id: "verify@2", node_id: "verify", properties: { text: "second visit prompt" }, }), - makeEnvelope({ - seq: 3, + envelope(3, { event: "agent.message", stage_id: "verify@1", node_id: "verify", properties: { text: "first visit reply" }, }), - makeEnvelope({ - seq: 4, + envelope(4, { event: "agent.message", stage_id: "verify@2", node_id: "verify", @@ -61,30 +57,61 @@ describe("turnsFromEvents", () => { }), ]; - const firstVisit = turnsFromEvents(events, "verify@1"); + const firstVisit = eventsToActivity(events, "verify@1"); expect(firstVisit).toEqual([ { kind: "system", content: "first visit prompt" }, { kind: "assistant", content: "first visit reply" }, ]); - const secondVisit = turnsFromEvents(events, "verify@2"); + const secondVisit = eventsToActivity(events, "verify@2"); expect(secondVisit).toEqual([ { kind: "system", content: "second visit prompt" }, { kind: "assistant", content: "second visit reply" }, ]); }); + test("pairs command.started + command.completed into a single command turn", () => { + const events: EventEnvelope[] = [ + envelope(1, { + event: "command.started", + node_id: "fmt", + properties: { script: "cargo fmt", language: "shell" }, + }), + envelope(2, { + event: "command.completed", + node_id: "fmt", + properties: { + stdout: "ok", + stderr: "", + exit_code: 0, + duration_ms: 12, + termination: "exited", + }, + }), + ]; + + const turns = eventsToActivity(events, "fmt"); + expect(turns).toHaveLength(1); + expect(turns[0]).toMatchObject({ + kind: "command", + stageId: "fmt", + script: "cargo fmt", + language: "shell", + stdout: "ok", + exitCode: 0, + running: false, + }); + }); + test("command turn carries the requested stage_id, no @1 fallback", () => { const events: EventEnvelope[] = [ - makeEnvelope({ - seq: 1, + envelope(1, { event: "command.started", stage_id: "verify@2", node_id: "verify", properties: { script: "echo hi", language: "shell" }, }), - makeEnvelope({ - seq: 2, + envelope(2, { event: "command.completed", stage_id: "verify@2", node_id: "verify", @@ -98,7 +125,7 @@ describe("turnsFromEvents", () => { }), ]; - const turns = turnsFromEvents(events, "verify@2"); + const turns = eventsToActivity(events, "verify@2"); expect(turns).toHaveLength(1); const turn = turns[0]; expect(turn.kind).toBe("command"); @@ -108,4 +135,72 @@ describe("turnsFromEvents", () => { expect(turn.running).toBe(false); } }); + + test("pairs agent.tool.started + agent.tool.completed into a single tool turn", () => { + const events: EventEnvelope[] = [ + envelope(1, { + event: "agent.tool.started", + node_id: "detect-drift", + properties: { + tool_call_id: "call-1", + tool_name: "read_file", + arguments: { path: "config.toml" }, + }, + }), + envelope(2, { + event: "agent.tool.completed", + node_id: "detect-drift", + properties: { + tool_call_id: "call-1", + tool_name: "read_file", + output: "[redis]", + is_error: false, + }, + }), + ]; + + const turns = eventsToActivity(events, "detect-drift"); + expect(turns).toHaveLength(1); + expect(turns[0].kind).toBe("tool"); + if (turns[0].kind === "tool") { + expect(turns[0].tools).toHaveLength(1); + expect(turns[0].tools[0]).toMatchObject({ + id: "call-1", + toolName: "read_file", + result: "[redis]", + isError: false, + }); + } + }); + + test("ignores unknown event types and events for other stages", () => { + const events: EventEnvelope[] = [ + envelope(1, { + 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: {}, + }), + envelope(4, { + event: "agent.message", + node_id: "other-stage", + properties: { text: "wrong stage" }, + }), + ]; + + const turns = eventsToActivity(events, "detect-drift"); + expect(turns).toHaveLength(1); + if (turns[0].kind === "assistant") { + expect(turns[0].content).toBe("signal"); + } + }); }); diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index 1176cb71a..a8fc69168 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/apps/fabro-web/app/routes/run-stages.tsx @@ -40,16 +40,14 @@ import type { Stage } from "../components/stage-sidebar"; import { EmptyState } from "../components/state"; import { CopyButton } from "../components/ui"; import { formatDurationSecs } from "../lib/format"; -import { fetchRunCommandLog, useRunEventsList, useRunStageTurns, useRunStages } from "../lib/queries"; +import { fetchRunCommandLog, useRunStageEvents, useRunStages } from "../lib/queries"; +import { STAGE_ACTIVITY_EVENT_TYPES, type StageActivityEventType } from "../lib/run-events"; import { ACTIVE_STAGE_STATES, formatStageLabel, mapRunStagesToSidebarStages } from "../lib/stage-sidebar"; import { getNumber, getString, type UnknownRecord } from "../lib/unknown"; import { CommandOutputStream, CommandTermination, type EventEnvelope, - type StageTurn as ApiStageTurn, - type PaginatedStageTurnList, - type PaginatedEventList, } from "@qltysh/fabro-api-client"; export const handle = { wide: true }; @@ -68,17 +66,40 @@ function readTermination(props: UnknownRecord): CommandTermination { return CommandTermination.EXITED; } -export function turnsFromEvents(events: EventEnvelope[], stageId: string): TurnType[] { - const stageEvents = events.filter((e) => e.stage_id === stageId); +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}`); +} + +function activityEventStageId(event: EventEnvelope): string | undefined { + if (typeof event.stage_id === "string") return event.stage_id; + if (typeof event.node_id === "string") return event.node_id; + return getString(event.properties ?? {}, "node_id"); +} + +export function eventsToActivity(events: EventEnvelope[], stageId: string): TurnType[] { 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) { + const eventName = e.event; + if ( + activityEventStageId(e) !== stageId || + !eventName || + !STAGE_ACTIVITY_EVENT_SET.has(eventName) + ) { + 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 = eventName 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; @@ -136,6 +157,8 @@ export function turnsFromEvents(events: EventEnvelope[], stageId: string): TurnT pendingCommand = undefined; break; } + default: + assertNever(eventType); } } @@ -153,41 +176,6 @@ export function turnsFromEvents(events: EventEnvelope[], stageId: string): TurnT return turns; } -function mapApiStageTurn(t: ApiStageTurn): TurnType { - switch (t.kind) { - case "tool": - return { - kind: "tool", - tools: (t.tools ?? []).map((tu) => ({ - id: tu.id, - toolName: tu.tool_name, - input: tu.input, - result: tu.result, - isError: tu.is_error, - durationMs: tu.duration_ms, - })), - }; - case "system": - case "assistant": - return { kind: t.kind, content: t.content ?? "" }; - } -} - -function mapTurns( - turnsResult: PaginatedStageTurnList | null | undefined, - eventsResult: PaginatedEventList | null | undefined, - selectedStageId: string | undefined, -): TurnType[] { - if (!selectedStageId) return []; - if (turnsResult?.data?.length) { - return turnsResult.data.map(mapApiStageTurn); - } - if (eventsResult?.data) { - return turnsFromEvents(eventsResult.data, selectedStageId); - } - return []; -} - function Markdown({ content }: { content: string }) { const html = useMemo(() => markedSafe.parse(content, { async: false }) as string, [content]); return ( @@ -605,14 +593,14 @@ export default function RunStages() { ); const selectedStage = stages.find((s: Stage) => s.id === stageId) ?? stages[0]; - const turnsQuery = useRunStageTurns(id, selectedStage?.id); - const hasStageTurns = (turnsQuery.data?.data.length ?? 0) > 0; - const shouldLoadEventFallback = - !!selectedStage?.id && !turnsQuery.isLoading && !turnsQuery.error && !hasStageTurns; - const eventsQuery = useRunEventsList(id, shouldLoadEventFallback); + const selectedStageId = selectedStage?.id; + const stageEventsQuery = useRunStageEvents(id, selectedStageId); const turns = useMemo( - () => mapTurns(turnsQuery.data, eventsQuery.data, selectedStage?.id), - [eventsQuery.data, selectedStage?.id, turnsQuery.data], + () => + selectedStageId + ? eventsToActivity(stageEventsQuery.data ?? [], selectedStageId) + : [], + [stageEventsQuery.data, selectedStageId], ); const isActive = selectedStage ? ACTIVE_STAGE_STATES.has(selectedStage.status) : false; diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index a767a7262..4319f6774 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -1906,26 +1906,26 @@ paths: schema: $ref: "#/components/schemas/ErrorResponse" - /api/v1/runs/{id}/stages/{stageId}/turns: + /api/v1/runs/{id}/stages/{stageId}/events: get: - operationId: listStageTurns + operationId: listStageEvents tags: [Run Internals] - summary: List Stage Turns - description: Returns a paginated list of conversation turns within a specific stage, including system prompts, assistant responses, and tool invocations. + summary: List Stage Events + description: Returns a paginated JSON list of stored run events scoped to a single stage visit. parameters: - $ref: "#/components/parameters/RunId" - $ref: "#/components/parameters/StageId" - - $ref: "#/components/parameters/PageLimit" - - $ref: "#/components/parameters/PageOffset" + - $ref: "#/components/parameters/SinceSeq" + - $ref: "#/components/parameters/EventLimit" responses: "200": - description: Paginated list of conversation turns + description: Paginated list of stage events content: application/json: schema: - $ref: "#/components/schemas/PaginatedStageTurnList" + $ref: "#/components/schemas/PaginatedEventList" "404": - description: Run or stage not found + description: Run not found. headers: x-request-id: $ref: "#/components/headers/XRequestId" @@ -3856,20 +3856,6 @@ components: meta: $ref: "#/components/schemas/PaginationMeta" - PaginatedStageTurnList: - description: Paginated list of stage turns. - type: object - required: - - data - - meta - properties: - data: - type: array - items: - $ref: "#/components/schemas/StageTurn" - meta: - $ref: "#/components/schemas/PaginationMeta" - PaginatedApiQuestionList: description: Paginated list of pending questions. type: object @@ -6348,103 +6334,6 @@ components: description: 1-based visit count; bumped each time the workflow re-enters this node. example: 2 - ToolUse: - description: A single tool invocation with its input, result, and execution metadata. - type: object - required: - - id - - tool_name - - input - - result - - is_error - properties: - id: - type: string - description: Unique identifier for this tool invocation. Enables correlation in parallel tool use. - example: toolu_01A09q90qw90lq917835lq9 - tool_name: - type: string - description: Name of the tool that was invoked. - example: read_file - input: - type: string - description: JSON-encoded input passed to the tool. - example: '{ "path": "src/routes/auth.ts" }' - result: - type: string - description: Output returned by the tool. Contains the error message when is_error is true. - example: 'import { Router } from "express";' - is_error: - type: boolean - description: Whether the tool invocation failed. When true, the result field contains the error message. - example: false - duration_ms: - type: integer - description: Wall-clock execution time of the tool invocation in milliseconds. - example: 142 - - StageTurn: - description: A single turn in a stage conversation — a system prompt, assistant response, or tool invocation block. - discriminator: - propertyName: kind - mapping: - system: "#/components/schemas/SystemStageTurn" - assistant: "#/components/schemas/AssistantStageTurn" - tool: "#/components/schemas/ToolStageTurn" - oneOf: - - $ref: "#/components/schemas/SystemStageTurn" - - $ref: "#/components/schemas/AssistantStageTurn" - - $ref: "#/components/schemas/ToolStageTurn" - - SystemStageTurn: - description: A system prompt turn that sets the stage's instructions. - type: object - required: - - kind - - content - properties: - kind: - type: string - enum: [system] - content: - type: string - description: System prompt text. - example: You are a drift detection agent. Compare the production and staging environments. - - AssistantStageTurn: - description: An assistant response turn within a stage. - type: object - required: - - kind - - content - properties: - kind: - type: string - enum: [assistant] - content: - type: string - description: Assistant response text. - example: I'll start by loading the environment configurations for both production and staging. - - ToolStageTurn: - description: A tool invocation turn containing one or more tool calls. - type: object - required: - - kind - - tools - properties: - kind: - type: string - enum: [tool] - content: - type: string - description: Text accompanying the tool invocations, or null when the turn contains only tool calls. - tools: - type: array - description: Tool invocations executed in this turn. - items: - $ref: "#/components/schemas/ToolUse" - # ── File Diff Schemas ────────────────────────────────────────────── FileCheckpoint: diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index bc527007d..c900c8b91 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -15,15 +15,16 @@ use axum::http::StatusCode; use axum::response::sse::{Event, Sse}; use axum::response::{IntoResponse, Response}; use fabro_api::types::{ - CreateSecretRequest, DeleteSecretRequest, DiffFile, DiffStats, FileDiff, FileDiffChangeKind, - PaginatedRunFileList, RunArtifactListResponse, RunFilesMeta, + CreateSecretRequest, DeleteSecretRequest, DiffFile, DiffStats, EventEnvelope, FileDiff, + FileDiffChangeKind, PaginatedEventList, PaginatedRunFileList, PaginationMeta, + RunArtifactListResponse, RunFilesMeta, }; 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, parse_stage_id_path}; fn paginated_response( items: Vec, @@ -133,13 +134,39 @@ pub(crate) async fn get_run_stages( paginated_response(runs::stages(), &pagination) } -pub(crate) async fn get_stage_turns( +pub(crate) async fn get_stage_events( _auth: RequiredUser, State(_state): State>, - Path((_id, _stage_id)): Path<(String, String)>, - Query(pagination): Query, + Path((_id, stage_id)): Path<(String, String)>, + Query(params): Query, ) -> Response { - paginated_response(runs::turns(), &pagination) + let stage_id = match parse_stage_id_path(&stage_id) { + Ok(stage_id) => stage_id, + Err(response) => return response, + }; + let since_seq = params.since_seq(); + let limit = params.limit(); + let mut matches: Vec = runs::stage_events() + .into_iter() + .filter(|envelope| { + envelope.seq >= since_seq + && (envelope.event.stage_id.as_ref() == Some(&stage_id) + || (envelope.event.stage_id.is_none() + && stage_id.visit() == 1 + && envelope.event.node_id.as_deref() == Some(stage_id.node_id()))) + }) + .take(limit + 1) + .collect(); + let has_more = matches.len() > limit; + matches.truncate(limit); + ( + StatusCode::OK, + Json(PaginatedEventList { + data: matches, + meta: PaginationMeta { has_more }, + }), + ) + .into_response() } pub(crate) async fn list_run_artifacts_stub( @@ -1215,18 +1242,114 @@ mod runs { ] } - pub(super) fn turns() -> Vec { + pub(super) fn stage_events() -> Vec { + use fabro_model::BilledTokenCounts; + use fabro_types::run_event::agent::{ + AgentMessageProps, AgentToolCompletedProps, AgentToolStartedProps, + }; + use fabro_types::run_event::stage::StagePromptProps; + use fabro_types::{EventBody, EventEnvelope, RunEvent}; + + let run_id = demo_run_id(1); + let node_id = "detect-drift"; + let stage_id = fabro_types::StageId::new(node_id, 1); + let ts = ts("2026-03-06T14:30:00Z"); + + let make_envelope = |seq: u32, id: &str, body: EventBody| EventEnvelope { + seq, + event: RunEvent { + id: id.into(), + ts, + run_id, + node_id: Some(node_id.into()), + node_label: Some("Detect Drift".into()), + stage_id: Some(stage_id.clone()), + parallel_group_id: None, + parallel_branch_id: None, + session_id: None, + parent_session_id: None, + tool_call_id: None, + actor: None, + body, + }, + }; + vec![ - StageTurn::SystemStageTurn(SystemStageTurn { kind: SystemStageTurnKind::System, content: "You are a drift detection agent. Compare the production and staging environments and identify any configuration or code drift.".into() }), - StageTurn::AssistantStageTurn(AssistantStageTurn { kind: AssistantStageTurnKind::Assistant, content: "I'll start by loading the environment configurations for both production and staging to compare them.".into() }), - StageTurn::ToolStageTurn(ToolStageTurn { - kind: ToolStageTurnKind::Tool, content: None, - tools: vec![ - ToolUse { id: "toolu_01".into(), tool_name: "read_file".into(), input: r#"{ "path": "environments/production/config.toml" }"#.into(), result: "[redis]\nhost = \"redis-prod.internal\"\nport = 6379".into(), is_error: false, duration_ms: Some(45) }, - ToolUse { id: "toolu_02".into(), tool_name: "read_file".into(), input: r#"{ "path": "environments/staging/config.toml" }"#.into(), result: "[redis]\nhost = \"redis-staging.internal\"\nport = 6379".into(), is_error: false, duration_ms: Some(38) }, - ], - }), - StageTurn::AssistantStageTurn(AssistantStageTurn { kind: AssistantStageTurnKind::Assistant, content: "I've detected drift in 3 resources between production and staging:\n\n1. **redis.max_connections** — production has 200, staging has 100\n2. **redis.tls** — enabled in production, disabled in staging\n3. **iam.session_duration** — production uses 3600s, staging uses 1800s".into() }), + make_envelope( + 1, + "evt-detect-drift-1", + EventBody::StagePrompt(StagePromptProps { + visit: 1, + text: "You are a drift detection agent. Compare the production and staging environments and identify any configuration or code drift.".into(), + mode: None, + provider: None, + model: None, + }), + ), + make_envelope( + 2, + "evt-detect-drift-2", + EventBody::AgentMessage(AgentMessageProps { + text: "I'll start by loading the environment configurations for both production and staging to compare them.".into(), + model: "Opus 4.6".into(), + billing: BilledTokenCounts::default(), + tool_call_count: 0, + visit: 1, + }), + ), + make_envelope( + 3, + "evt-detect-drift-3", + EventBody::AgentToolStarted(AgentToolStartedProps { + tool_name: "read_file".into(), + tool_call_id: "toolu_01".into(), + arguments: serde_json::json!({ "path": "environments/production/config.toml" }), + visit: 1, + }), + ), + make_envelope( + 4, + "evt-detect-drift-4", + EventBody::AgentToolCompleted(AgentToolCompletedProps { + tool_name: "read_file".into(), + tool_call_id: "toolu_01".into(), + output: serde_json::json!("[redis]\nhost = \"redis-prod.internal\"\nport = 6379"), + is_error: false, + visit: 1, + }), + ), + make_envelope( + 5, + "evt-detect-drift-5", + EventBody::AgentToolStarted(AgentToolStartedProps { + tool_name: "read_file".into(), + tool_call_id: "toolu_02".into(), + arguments: serde_json::json!({ "path": "environments/staging/config.toml" }), + visit: 1, + }), + ), + make_envelope( + 6, + "evt-detect-drift-6", + EventBody::AgentToolCompleted(AgentToolCompletedProps { + tool_name: "read_file".into(), + tool_call_id: "toolu_02".into(), + output: serde_json::json!("[redis]\nhost = \"redis-staging.internal\"\nport = 6379"), + is_error: false, + visit: 1, + }), + ), + make_envelope( + 7, + "evt-detect-drift-7", + EventBody::AgentMessage(AgentMessageProps { + text: "I've detected drift in 3 resources between production and staging:\n\n1. **redis.max_connections** — production has 200, staging has 100\n2. **redis.tls** — enabled in production, disabled in staging\n3. **iam.session_duration** — production uses 3600s, staging uses 1800s".into(), + model: "Opus 4.6".into(), + billing: BilledTokenCounts::default(), + tool_call_count: 0, + visit: 1, + }), + ), ] } diff --git a/lib/crates/fabro-server/src/principal_middleware.rs b/lib/crates/fabro-server/src/principal_middleware.rs index e8971831c..822d87dc6 100644 --- a/lib/crates/fabro-server/src/principal_middleware.rs +++ b/lib/crates/fabro-server/src/principal_middleware.rs @@ -52,6 +52,7 @@ pub(crate) struct RequestAuth(pub(crate) AuthContextSlot); pub(crate) struct RequiredUser(pub(crate) UserPrincipal); pub(crate) struct RequireRunScoped(pub(crate) RunId); pub(crate) struct RequireRunBlob(pub(crate) RunId, pub(crate) RunBlobId); +pub(crate) struct RequireRunStageScoped(pub(crate) RunId, pub(crate) String); pub(crate) struct RequireStageArtifact(pub(crate) RunId, pub(crate) StageId); pub(crate) struct RequireCommandLog( pub(crate) RunId, @@ -203,6 +204,23 @@ impl FromRequestParts> for RequireRunBlob { } } +impl FromRequestParts> for RequireRunStageScoped { + type Rejection = Response; + + async fn from_request_parts( + parts: &mut Parts, + state: &Arc, + ) -> Result { + let Path((id, stage_id)): Path<(String, String)> = Path::from_request_parts(parts, state) + .await + .map_err(IntoResponse::into_response)?; + let run_id = parse_run_id_path(&id)?; + require_worker_or_user_for_run(&auth_slot_from_parts(parts), &run_id) + .map_err(IntoResponse::into_response)?; + Ok(Self(run_id, stage_id)) + } +} + impl FromRequestParts> for RequireStageArtifact { type Rejection = Response; diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 26fbdd8d6..965989c8c 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -125,7 +125,7 @@ use crate::ip_allowlist::{IpAllowlistConfig, ip_allowlist_middleware}; use crate::jwt_auth::{self, AuthMode}; use crate::principal_middleware::{ AuthContextSlot, RequestAuth, RequestAuthContext, RequireRunBlob, RequireRunScoped, - RequireStageArtifact, RequiredUser, principal_middleware, + RequireRunStageScoped, RequireStageArtifact, RequiredUser, principal_middleware, }; use crate::request_id::{self, RequestId}; use crate::run_files::{FilesInFlight, new_files_in_flight}; @@ -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 ff3889cd5..fce4a4d11 100644 --- a/lib/crates/fabro-server/src/server/handler/events.rs +++ b/lib/crates/fabro-server/src/server/handler/events.rs @@ -3,9 +3,10 @@ use std::sync::Arc; use super::super::{ ApiError, AppState, AppendEventResponse, BroadcastStream, Event, EventBody, EventEnvelope, EventPayload, HashSet, IntoResponse, Json, KeepAlive, PaginatedEventList, PaginationMeta, Path, - Query, RequireRunScoped, RequiredUser, Response, Router, RunEvent, RunId, RunStatus, Sse, - State, StatusCode, StreamExt, UnboundedReceiverStream, broadcast, get, mpsc, parse_run_id_path, - redact_jsonl_line, reject_if_archived, update_live_run_from_event, + Query, RequireRunScoped, RequireRunStageScoped, RequiredUser, Response, Router, RunEvent, + RunId, RunStatus, Sse, State, StatusCode, StreamExt, UnboundedReceiverStream, broadcast, get, + mpsc, parse_run_id_path, parse_stage_id_path, redact_jsonl_line, reject_if_archived, + update_live_run_from_event, }; pub(super) fn routes() -> Router> { @@ -15,11 +16,15 @@ pub(super) fn routes() -> Router> { "/runs/{id}/events", get(list_run_events).post(append_run_event), ) + .route( + "/runs/{id}/stages/{stageId}/events", + get(list_run_stage_events), + ) .route("/runs/{id}/attach", get(attach_run_events)) } #[derive(serde::Deserialize)] -struct EventListParams { +pub(crate) struct EventListParams { #[serde(default)] since_seq: Option, #[serde(default)] @@ -27,11 +32,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) } } @@ -200,6 +205,39 @@ async fn list_run_events( } } +async fn list_run_stage_events( + RequireRunStageScoped(id, stage_id): RequireRunStageScoped, + State(state): State>, + Query(params): Query, +) -> Response { + let stage_id = match parse_stage_id_path(&stage_id) { + Ok(stage_id) => stage_id, + Err(response) => return response, + }; + let since_seq = params.since_seq(); + let limit = params.limit(); + match state.store.open_run_reader(&id).await { + Ok(run_store) => match run_store + .list_events_for_stage_from_with_limit(&stage_id, since_seq, limit) + .await + { + Ok(mut events) => { + let has_more = events.len() > limit; + events.truncate(limit); + Json(PaginatedEventList { + data: events, + meta: PaginationMeta { has_more }, + }) + .into_response() + } + Err(err) => { + ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } + }, + Err(_) => ApiError::not_found("Run not found.").into_response(), + } +} + async fn attach_run_events( _auth: RequiredUser, State(state): State>, @@ -342,3 +380,264 @@ fn denied_lifecycle_event_name(body: &EventBody) -> Option<&'static str> { _ => None, } } + +#[cfg(test)] +mod stage_events_tests { + use axum::body::{Body, to_bytes}; + use axum::http::{Request, StatusCode, header}; + use fabro_store::EventPayload; + use fabro_types::RunId; + use serde_json::json; + use tower::ServiceExt; + + use crate::test_support::{build_test_router, test_app_state}; + + fn req_get(uri: &str) -> Request { + Request::builder() + .method("GET") + .uri(uri) + .body(Body::empty()) + .expect("stage events GET request should build") + } + + fn make_event(run_id: &RunId, idx: u32, node_id: Option<&str>) -> EventPayload { + make_event_with_stage_id(run_id, idx, node_id, None) + } + + fn make_event_with_stage_id( + run_id: &RunId, + idx: u32, + node_id: Option<&str>, + stage_id: Option<&str>, + ) -> EventPayload { + let mut value = json!({ + "id": format!("evt-{idx}"), + "ts": "2026-04-09T12:00:00Z", + "run_id": run_id.to_string(), + "event": "stage.prompt", + "properties": { + "visit": 1, + "text": format!("prompt {idx}"), + }, + }); + if let Some(node) = node_id { + value + .as_object_mut() + .unwrap() + .insert("node_id".into(), json!(node)); + } + if let Some(stage_id) = stage_id { + value + .as_object_mut() + .unwrap() + .insert("stage_id".into(), json!(stage_id)); + } + EventPayload::new(value, run_id).expect("event payload should validate") + } + + async fn body_json(response: axum::response::Response) -> serde_json::Value { + let bytes = to_bytes(response.into_body(), usize::MAX) + .await + .expect("response body should fit in memory"); + serde_json::from_slice(&bytes).expect("response body should be valid JSON") + } + + async fn seed_run_with_mixed_events() -> (RunId, axum::Router) { + let state = test_app_state(); + let app = build_test_router(state.clone()); + let run_id = RunId::new(); + let run_store = state + .store_ref() + .create_run(&run_id) + .await + .expect("test run should be creatable"); + + // Seed 200 unrelated 'beta' events first so any node-blind + // truncation would lose the sparse 'alpha' tail. Then 3 'alpha' + // events past seq 100, plus a couple with no node_id at all. + for idx in 1..=200_u32 { + run_store + .append_event(&make_event(&run_id, idx, Some("beta"))) + .await + .expect("append should succeed"); + } + run_store + .append_event(&make_event(&run_id, 201, None)) + .await + .expect("append should succeed"); + for idx in 202..=204_u32 { + run_store + .append_event(&make_event(&run_id, idx, Some("alpha"))) + .await + .expect("append should succeed"); + } + + (run_id, app) + } + + #[tokio::test] + async fn returns_only_matching_node_events_in_seq_order() { + let (run_id, app) = seed_run_with_mixed_events().await; + let response = app + .oneshot(req_get(&format!( + "/api/v1/runs/{run_id}/stages/alpha@1/events" + ))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = body_json(response).await; + let data = body["data"].as_array().expect("data is array"); + let seqs: Vec = data.iter().map(|e| e["seq"].as_u64().unwrap()).collect(); + assert_eq!(seqs, vec![202, 203, 204]); + assert_eq!(body["meta"]["has_more"], false); + } + + #[tokio::test] + async fn since_seq_filters_to_events_with_seq_at_least_k() { + let (run_id, app) = seed_run_with_mixed_events().await; + let response = app + .oneshot(req_get(&format!( + "/api/v1/runs/{run_id}/stages/alpha@1/events?since_seq=203" + ))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = body_json(response).await; + let seqs: Vec = body["data"] + .as_array() + .unwrap() + .iter() + .map(|e| e["seq"].as_u64().unwrap()) + .collect(); + assert_eq!(seqs, vec![203, 204]); + } + + #[tokio::test] + async fn limit_one_returns_first_envelope_with_has_more_true() { + let (run_id, app) = seed_run_with_mixed_events().await; + let response = app + .oneshot(req_get(&format!( + "/api/v1/runs/{run_id}/stages/alpha@1/events?limit=1" + ))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = body_json(response).await; + let data = body["data"].as_array().unwrap(); + assert_eq!(data.len(), 1); + assert_eq!(data[0]["seq"].as_u64().unwrap(), 202); + assert_eq!(body["meta"]["has_more"], true); + } + + #[tokio::test] + async fn unknown_stage_in_existing_run_returns_empty_list_with_no_more() { + let (run_id, app) = seed_run_with_mixed_events().await; + let response = app + .oneshot(req_get(&format!( + "/api/v1/runs/{run_id}/stages/unknown-stage@1/events" + ))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = body_json(response).await; + assert_eq!(body["data"].as_array().unwrap().len(), 0); + assert_eq!(body["meta"]["has_more"], false); + } + + #[tokio::test] + async fn missing_run_returns_404_with_run_not_found() { + let app = build_test_router(test_app_state()); + // A syntactically valid RunId that the store has never seen, so + // `parse_run_id_path` succeeds but `open_run_reader` fails — that + // exercises the handler's not-found branch rather than the path + // parser's 400 branch. + let absent = RunId::new(); + let response = app + .oneshot(req_get(&format!( + "/api/v1/runs/{absent}/stages/alpha@1/events" + ))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + + let body = body_json(response).await; + let detail = body["errors"][0]["detail"] + .as_str() + .expect("error detail string"); + assert!( + detail.contains("Run not found."), + "unexpected error body: {body}" + ); + } + + #[tokio::test] + async fn unauthenticated_request_is_rejected() { + let state = test_app_state(); + // Bypass `build_test_router`'s auto-injected bearer token by + // building the raw router directly. The principal middleware sees + // a missing Authorization header and the extractor enforces auth. + let app = crate::server::build_router(state, crate::test_support::test_auth_mode()); + let run_id = RunId::new(); + + let request = Request::builder() + .method("GET") + .uri(format!("/api/v1/runs/{run_id}/stages/alpha@1/events")) + .header(header::ACCEPT, "application/json") + .body(Body::empty()) + .unwrap(); + let response = app.oneshot(request).await.unwrap(); + + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + + #[tokio::test] + async fn returns_only_requested_visit_when_stage_id_is_present() { + let state = test_app_state(); + let app = build_test_router(state.clone()); + let run_id = RunId::new(); + let run_store = state + .store_ref() + .create_run(&run_id) + .await + .expect("test run should be creatable"); + run_store + .append_event(&make_event_with_stage_id( + &run_id, + 1, + Some("verify"), + Some("verify@1"), + )) + .await + .expect("append should succeed"); + run_store + .append_event(&make_event_with_stage_id( + &run_id, + 2, + Some("verify"), + Some("verify@2"), + )) + .await + .expect("append should succeed"); + + let response = app + .oneshot(req_get(&format!( + "/api/v1/runs/{run_id}/stages/verify@2/events" + ))) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = body_json(response).await; + let seqs: Vec = body["data"] + .as_array() + .unwrap() + .iter() + .map(|e| e["seq"].as_u64().unwrap()) + .collect(); + assert_eq!(seqs, vec![2]); + } +} diff --git a/lib/crates/fabro-server/src/server/handler/mod.rs b/lib/crates/fabro-server/src/server/handler/mod.rs index d2625395b..79a3d8246 100644 --- a/lib/crates/fabro-server/src/server/handler/mod.rs +++ b/lib/crates/fabro-server/src/server/handler/mod.rs @@ -57,8 +57,8 @@ pub(super) fn demo_routes() -> Router> { .route("/runs/{id}/artifacts", get(demo::list_run_artifacts_stub)) .route("/runs/{id}/files", get(demo::list_run_files_stub)) .route( - "/runs/{id}/stages/{stageId}/turns", - get(demo::get_stage_turns), + "/runs/{id}/stages/{stageId}/events", + get(demo::get_stage_events), ) .route( "/runs/{id}/stages/{stageId}/artifacts", @@ -113,7 +113,6 @@ pub(super) fn demo_routes() -> Router> { pub(super) fn real_routes() -> Router> { Router::new() - .route("/runs/{id}/stages/{stageId}/turns", get(not_implemented)) .route("/runs/{id}/steer", post(not_implemented)) .route("/workflows", get(not_implemented)) .route("/workflows/{name}", get(not_implemented)) diff --git a/lib/crates/fabro-server/tests/it/event_pagination.rs b/lib/crates/fabro-server/tests/it/event_pagination.rs new file mode 100644 index 000000000..cd6ceec94 --- /dev/null +++ b/lib/crates/fabro-server/tests/it/event_pagination.rs @@ -0,0 +1,75 @@ +//! Cursor-pagination tests for the per-stage events endpoint (demo mode). +//! +//! The stage-events route uses `since_seq=` + `limit=` (cursor-based) instead +//! of the offset-based `page[limit]/page[offset]` pagination used by other +//! list endpoints, so it gets its own test rather than living in the generic +//! offset-shape matrix. + +#![allow( + clippy::absolute_paths, + reason = "This test module prefers explicit type paths over extra imports." +)] + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use tower::ServiceExt; + +use super::helpers::{response_json, test_app_state}; + +async fn get_json(app: &axum::Router, uri: &str) -> serde_json::Value { + let req = Request::builder() + .method("GET") + .uri(uri) + .header("x-fabro-demo", "1") + .body(Body::empty()) + .expect("event pagination request should build"); + let response = app.clone().oneshot(req).await.unwrap(); + response_json(response, StatusCode::OK, format!("GET {uri}")).await +} + +#[tokio::test] +async fn demo_stage_events_default_returns_all_fixture_events_with_no_more() { + let app = fabro_server::test_support::build_test_router(test_app_state()); + + let body = get_json(&app, "/api/v1/runs/run-1/stages/detect-drift@1/events").await; + let data = body["data"].as_array().expect("data is an array"); + + assert_eq!(data.len(), 7, "all seven fixture events should be returned"); + assert_eq!(body["meta"]["has_more"], false); +} + +#[tokio::test] +async fn demo_stage_events_limit_one_signals_has_more() { + let app = fabro_server::test_support::build_test_router(test_app_state()); + + let body = get_json( + &app, + "/api/v1/runs/run-1/stages/detect-drift@1/events?limit=1", + ) + .await; + let data = body["data"].as_array().expect("data is an array"); + + assert_eq!(data.len(), 1); + assert_eq!(body["meta"]["has_more"], true); +} + +#[tokio::test] +async fn demo_stage_events_since_seq_filters_out_earlier_events() { + let app = fabro_server::test_support::build_test_router(test_app_state()); + + // The fixture seqs are 1..=7. since_seq=4 should skip the first three. + let body = get_json( + &app, + "/api/v1/runs/run-1/stages/detect-drift@1/events?since_seq=4", + ) + .await; + let data = body["data"].as_array().expect("data is an array"); + + assert_eq!(data.len(), 4); + let seqs: Vec = data + .iter() + .map(|envelope| envelope["seq"].as_u64().expect("seq is a number")) + .collect(); + assert_eq!(seqs, vec![4, 5, 6, 7]); + assert_eq!(body["meta"]["has_more"], false); +} diff --git a/lib/crates/fabro-server/tests/it/main.rs b/lib/crates/fabro-server/tests/it/main.rs index 9e5ddab9f..865bb8132 100644 --- a/lib/crates/fabro-server/tests/it/main.rs +++ b/lib/crates/fabro-server/tests/it/main.rs @@ -4,6 +4,7 @@ )] mod api; +mod event_pagination; mod helpers; mod openapi_conformance; mod pagination; diff --git a/lib/crates/fabro-server/tests/it/pagination.rs b/lib/crates/fabro-server/tests/it/pagination.rs index 64174fd78..f773ac207 100644 --- a/lib/crates/fabro-server/tests/it/pagination.rs +++ b/lib/crates/fabro-server/tests/it/pagination.rs @@ -56,10 +56,6 @@ const ENDPOINTS: &[PaginatedEndpoint] = &[ path: "/api/v1/models", name: "listModels", }, - PaginatedEndpoint { - path: "/api/v1/runs/run-1/stages/detect-drift/turns", - name: "listStageTurns", - }, PaginatedEndpoint { path: "/api/v1/runs/run-1/questions", name: "listRunQuestions", diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index a334281b9..f28e52341 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -12,7 +12,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use super::blob_store::BlobStore; use crate::run_state::{EventProjectionCache, RunProjectionReducer, build_summary}; -use crate::{Error, EventEnvelope, EventPayload, Result, RunProjection, keys}; +use crate::{Error, EventEnvelope, EventPayload, Result, RunProjection, StageId, keys}; const DEFAULT_EVENT_TAIL_LIMIT: usize = 1024; #[derive(Clone)] @@ -213,6 +213,30 @@ impl RunDatabase { list_events_from_with_limit(&self.inner.db, &self.inner.run_id, start_seq, limit).await } + /// Returns up to `limit + 1` events for the given stage visit, + /// starting at `start_seq`. The `+1` lets callers compute `has_more`. + /// + /// Implementation note: scans the unbounded run-event prefix and + /// filters by stage identity *before* applying `limit`, so a stage with + /// matches sparsely scattered late in the event log still returns its + /// full slice (no premature truncation from a generic `limit`-bounded + /// scan). + pub async fn list_events_for_stage_from_with_limit( + &self, + stage_id: &StageId, + start_seq: u32, + limit: usize, + ) -> Result> { + list_events_for_stage_from_with_limit( + &self.inner.db, + &self.inner.run_id, + stage_id, + start_seq, + limit, + ) + .await + } + pub fn watch_events_from( &self, seq: u32, @@ -348,6 +372,74 @@ where Ok(events) } +async fn list_events_for_stage_from_with_limit( + db: &R, + run_id: &RunId, + stage_id: &StageId, + start_seq: u32, + limit: usize, +) -> Result> +where + R: DbRead + Sync, +{ + // Unbounded scan first: filtering by stage identity 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 stage identity fields 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 StageIdProbe<'a> { + #[serde(default, borrow)] + stage_id: Option<&'a str>, + #[serde(default, borrow)] + node_id: Option<&'a str>, + } + + let stage_id_string = stage_id.to_string(); + 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? { + let key = key_to_string(&entry.key)?; + let Some(seq) = keys::parse_event_seq(&key) else { + continue; + }; + if seq < start_seq { + continue; + } + let probe: StageIdProbe = serde_json::from_slice(&entry.value)?; + let matches_stage_id = probe.stage_id == Some(stage_id_string.as_str()); + let matches_legacy_node_id = probe.stage_id.is_none() + && stage_id.visit() == 1 + && probe.node_id == Some(stage_id.node_id()); + if !matches_stage_id && !matches_legacy_node_id { + continue; + } + let event: RunEvent = serde_json::from_slice(&entry.value)?; + 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); + Ok(events) +} + async fn list_blobs(db: &R) -> Result> where R: DbRead + Sync, @@ -375,9 +467,12 @@ mod tests { use std::sync::Arc; use std::time::Duration; + use fabro_types::{RunId, StageId}; use object_store::memory::InMemory; + use serde_json::json; + + use crate::{Database, EventPayload}; - use crate::Database; #[tokio::test] async fn list_blobs_reads_global_cas_namespace() { let object_store = Arc::new(InMemory::new()); @@ -394,4 +489,190 @@ mod tests { assert_eq!(blob_ids, vec![first_id, second_id]); } + + fn stage_prompt_payload(run_id: &RunId, idx: u32, node_id: Option<&str>) -> EventPayload { + stage_prompt_payload_for_stage(run_id, idx, node_id, None) + } + + fn stage_prompt_payload_for_stage( + run_id: &RunId, + idx: u32, + node_id: Option<&str>, + stage_id: Option<&StageId>, + ) -> EventPayload { + let mut value = json!({ + "id": format!("evt-{idx}"), + "ts": "2026-04-09T12:00:00Z", + "run_id": run_id.to_string(), + "event": "stage.prompt", + "properties": { + "visit": 1, + "text": format!("prompt {idx}"), + }, + }); + if let Some(node_id) = node_id { + value + .as_object_mut() + .unwrap() + .insert("node_id".into(), json!(node_id)); + } + if let Some(stage_id) = stage_id { + value + .as_object_mut() + .unwrap() + .insert("stage_id".into(), json!(stage_id.to_string())); + } + EventPayload::new(value, run_id).unwrap() + } + + async fn fresh_run() -> super::RunDatabase { + let object_store = Arc::new(InMemory::new()); + let store = Database::new(object_store, "", Duration::from_millis(1), None); + let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap(); + store.create_run(&run_id).await.unwrap() + } + + #[tokio::test] + async fn list_events_for_stage_returns_only_matching_events_in_seq_order() { + let run = fresh_run().await; + let run_id = run.run_id(); + run.append_event(&stage_prompt_payload(&run_id, 1, Some("alpha"))) + .await + .unwrap(); + run.append_event(&stage_prompt_payload(&run_id, 2, Some("beta"))) + .await + .unwrap(); + run.append_event(&stage_prompt_payload(&run_id, 3, Some("alpha"))) + .await + .unwrap(); + + let events = run + .list_events_for_stage_from_with_limit(&StageId::new("alpha", 1), 1, 100) + .await + .unwrap(); + + let seqs: Vec = events.iter().map(|e| e.seq).collect(); + assert_eq!(seqs, vec![1, 3]); + } + + #[tokio::test] + async fn list_events_for_stage_skips_events_with_no_stage_identity() { + let run = fresh_run().await; + let run_id = run.run_id(); + run.append_event(&stage_prompt_payload(&run_id, 1, None)) + .await + .unwrap(); + run.append_event(&stage_prompt_payload(&run_id, 2, Some("alpha"))) + .await + .unwrap(); + + let events = run + .list_events_for_stage_from_with_limit(&StageId::new("alpha", 1), 1, 100) + .await + .unwrap(); + + let seqs: Vec = events.iter().map(|e| e.seq).collect(); + assert_eq!(seqs, vec![2]); + } + + #[tokio::test] + async fn list_events_for_stage_paginates_via_start_seq_on_filtered_slice() { + let run = fresh_run().await; + let run_id = run.run_id(); + for idx in 1..=5 { + let node = if idx % 2 == 0 { "beta" } else { "alpha" }; + run.append_event(&stage_prompt_payload(&run_id, idx, Some(node))) + .await + .unwrap(); + } + + // alpha events live at seqs 1, 3, 5. Start at seq=2 should skip seq=1. + let events = run + .list_events_for_stage_from_with_limit(&StageId::new("alpha", 1), 2, 100) + .await + .unwrap(); + + let seqs: Vec = events.iter().map(|e| e.seq).collect(); + assert_eq!(seqs, vec![3, 5]); + } + + #[tokio::test] + async fn list_events_for_stage_walks_past_unrelated_events_for_sparse_matches() { + let run = fresh_run().await; + let run_id = run.run_id(); + // 200 unrelated events first. + for idx in 1..=200 { + run.append_event(&stage_prompt_payload(&run_id, idx, Some("noise"))) + .await + .unwrap(); + } + // Then 3 sparse "alpha" events at the tail. + for idx in 201..=203 { + run.append_event(&stage_prompt_payload(&run_id, idx, Some("alpha"))) + .await + .unwrap(); + } + + // limit smaller than the number of unrelated events would have + // truncated the upstream scan if we had post-filtered. + let events = run + .list_events_for_stage_from_with_limit(&StageId::new("alpha", 1), 1, 5) + .await + .unwrap(); + + let seqs: Vec = events.iter().map(|e| e.seq).collect(); + assert_eq!(seqs, vec![201, 202, 203]); + } + + #[tokio::test] + async fn list_events_for_stage_returns_limit_plus_one_for_has_more_signal() { + let run = fresh_run().await; + let run_id = run.run_id(); + for idx in 1..=5 { + run.append_event(&stage_prompt_payload(&run_id, idx, Some("alpha"))) + .await + .unwrap(); + } + + let events = run + .list_events_for_stage_from_with_limit(&StageId::new("alpha", 1), 1, 2) + .await + .unwrap(); + + // With limit=2, we expect up to limit+1 = 3 envelopes so the + // caller can compute has_more. + assert_eq!(events.len(), 3); + } + + #[tokio::test] + async fn list_events_for_stage_prefers_stage_id_over_node_id() { + let run = fresh_run().await; + let run_id = run.run_id(); + let first_visit = StageId::new("verify", 1); + let second_visit = StageId::new("verify", 2); + run.append_event(&stage_prompt_payload_for_stage( + &run_id, + 1, + Some("verify"), + Some(&first_visit), + )) + .await + .unwrap(); + run.append_event(&stage_prompt_payload_for_stage( + &run_id, + 2, + Some("verify"), + Some(&second_visit), + )) + .await + .unwrap(); + + let events = run + .list_events_for_stage_from_with_limit(&second_visit, 1, 100) + .await + .unwrap(); + + let seqs: Vec = events.iter().map(|e| e.seq).collect(); + assert_eq!(seqs, vec![2]); + } } diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index 3113a0298..2dffdba54 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -30,7 +30,6 @@ models/artifact-batch-upload-manifest.ts models/artifact-entry.ts models/artifact-list-response.ts models/artifacts-settings.ts -models/assistant-stage-turn.ts models/auth-method.ts models/billed-token-counts.ts models/billing-by-model.ts @@ -168,7 +167,6 @@ models/paginated-run-file-list.ts models/paginated-run-list.ts models/paginated-run-stage-list.ts models/paginated-saved-query-list.ts -models/paginated-stage-turn-list.ts models/pagination-meta.ts models/pending-interview-record.ts models/pre-run-push-outcome-failed.ts @@ -300,7 +298,6 @@ models/stage-completion.ts models/stage-outcome.ts models/stage-projection.ts models/stage-state.ts -models/stage-turn.ts models/start-run-request.ts models/submit-answer-request.ts models/success-reason.ts @@ -308,13 +305,10 @@ models/system-actor-kind.ts models/system-features.ts models/system-info-response.ts models/system-run-counts.ts -models/system-stage-turn.ts models/teams-integration-settings.ts models/terminal-status.ts models/timeline-entry-response.ts models/tls-mode.ts -models/tool-stage-turn.ts -models/tool-use.ts models/user-response.ts models/validate-response.ts models/webhook-strategy.ts diff --git a/lib/packages/fabro-api-client/src/api/run-internals-api.ts b/lib/packages/fabro-api-client/src/api/run-internals-api.ts index 0852b16f6..6520b0fc6 100644 --- a/lib/packages/fabro-api-client/src/api/run-internals-api.ts +++ b/lib/packages/fabro-api-client/src/api/run-internals-api.ts @@ -36,8 +36,6 @@ import type { PaginatedEventList } from '../models'; // @ts-ignore import type { PaginatedRunStageList } from '../models'; // @ts-ignore -import type { PaginatedStageTurnList } from '../models'; -// @ts-ignore import type { RunArtifactListResponse } from '../models'; // @ts-ignore import type { RunCheckpoint } from '../models'; @@ -527,21 +525,21 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config }; }, /** - * Returns a paginated list of conversation turns within a specific stage, including system prompts, assistant responses, and tool invocations. - * @summary List Stage Turns + * Returns a paginated JSON list of stored run events scoped to a single stage visit. + * @summary List Stage Events * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {number} [pageLimit] Maximum number of items to return per page. - * @param {number} [pageOffset] Number of items to skip before returning results. + * @param {number} [sinceSeq] First event sequence number to include. + * @param {number} [limit] Maximum number of events to return. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - listStageTurns: async (id: string, stageId: string, pageLimit?: number, pageOffset?: number, options: RawAxiosRequestConfig = {}): Promise => { + listStageEvents: async (id: string, stageId: string, sinceSeq?: number, limit?: number, options: RawAxiosRequestConfig = {}): Promise => { // verify required parameter 'id' is not null or undefined - assertParamExists('listStageTurns', 'id', id) + assertParamExists('listStageEvents', 'id', id) // verify required parameter 'stageId' is not null or undefined - assertParamExists('listStageTurns', 'stageId', stageId) - const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/turns` + assertParamExists('listStageEvents', 'stageId', stageId) + const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/events` .replace(`{${"id"}}`, encodeURIComponent(String(id))) .replace(`{${"stageId"}}`, encodeURIComponent(String(stageId))); // use dummy base URL string because the URL constructor only accepts absolute URLs. @@ -561,12 +559,12 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config // http bearer authentication required await setBearerAuthToObject(localVarHeaderParameter, configuration) - if (pageLimit !== undefined) { - localVarQueryParameter['page[limit]'] = pageLimit; + if (sinceSeq !== undefined) { + localVarQueryParameter['since_seq'] = sinceSeq; } - if (pageOffset !== undefined) { - localVarQueryParameter['page[offset]'] = pageOffset; + if (limit !== undefined) { + localVarQueryParameter['limit'] = limit; } localVarHeaderParameter['Accept'] = 'application/json'; @@ -964,19 +962,19 @@ export const RunInternalsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Returns a paginated list of conversation turns within a specific stage, including system prompts, assistant responses, and tool invocations. - * @summary List Stage Turns + * Returns a paginated JSON list of stored run events scoped to a single stage visit. + * @summary List Stage Events * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {number} [pageLimit] Maximum number of items to return per page. - * @param {number} [pageOffset] Number of items to skip before returning results. + * @param {number} [sinceSeq] First event sequence number to include. + * @param {number} [limit] Maximum number of events to return. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - async listStageTurns(id: string, stageId: string, pageLimit?: number, pageOffset?: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { - const localVarAxiosArgs = await localVarAxiosParamCreator.listStageTurns(id, stageId, pageLimit, pageOffset, options); + async listStageEvents(id: string, stageId: string, sinceSeq?: number, limit?: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.listStageEvents(id, stageId, sinceSeq, limit, options); const localVarOperationServerIndex = configuration?.serverIndex ?? 0; - const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.listStageTurns']?.[localVarOperationServerIndex]?.url; + const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.listStageEvents']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** @@ -1174,17 +1172,17 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b return localVarFp.listStageArtifacts(id, stageId, options).then((request) => request(axios, basePath)); }, /** - * Returns a paginated list of conversation turns within a specific stage, including system prompts, assistant responses, and tool invocations. - * @summary List Stage Turns + * Returns a paginated JSON list of stored run events scoped to a single stage visit. + * @summary List Stage Events * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {number} [pageLimit] Maximum number of items to return per page. - * @param {number} [pageOffset] Number of items to skip before returning results. + * @param {number} [sinceSeq] First event sequence number to include. + * @param {number} [limit] Maximum number of events to return. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - listStageTurns(id: string, stageId: string, pageLimit?: number, pageOffset?: number, options?: RawAxiosRequestConfig): AxiosPromise { - return localVarFp.listStageTurns(id, stageId, pageLimit, pageOffset, options).then((request) => request(axios, basePath)); + listStageEvents(id: string, stageId: string, sinceSeq?: number, limit?: number, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.listStageEvents(id, stageId, sinceSeq, limit, options).then((request) => request(axios, basePath)); }, /** * Uploads one or more artifacts for a stage. Intended for trusted internal callers. The server accepts both: - `application/octet-stream` for single-file uploads with the `filename` query parameter - strict manifest-first `multipart/form-data` uploads documented by `ArtifactBatchUploadManifest` The generated Rust client currently exposes the octet-stream variant because the OpenAPI code generator in this repo does not support multiple request media types on one operation. @@ -1374,17 +1372,17 @@ export class RunInternalsApi extends BaseAPI { } /** - * Returns a paginated list of conversation turns within a specific stage, including system prompts, assistant responses, and tool invocations. - * @summary List Stage Turns + * Returns a paginated JSON list of stored run events scoped to a single stage visit. + * @summary List Stage Events * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. - * @param {number} [pageLimit] Maximum number of items to return per page. - * @param {number} [pageOffset] Number of items to skip before returning results. + * @param {number} [sinceSeq] First event sequence number to include. + * @param {number} [limit] Maximum number of events to return. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - public listStageTurns(id: string, stageId: string, pageLimit?: number, pageOffset?: number, options?: RawAxiosRequestConfig) { - return RunInternalsApiFp(this.configuration).listStageTurns(id, stageId, pageLimit, pageOffset, options).then((request) => request(this.axios, this.basePath)); + public listStageEvents(id: string, stageId: string, sinceSeq?: number, limit?: number, options?: RawAxiosRequestConfig) { + return RunInternalsApiFp(this.configuration).listStageEvents(id, stageId, sinceSeq, limit, options).then((request) => request(this.axios, this.basePath)); } /** diff --git a/lib/packages/fabro-api-client/src/models/assistant-stage-turn.ts b/lib/packages/fabro-api-client/src/models/assistant-stage-turn.ts deleted file mode 100644 index 8e0a8edcf..000000000 --- a/lib/packages/fabro-api-client/src/models/assistant-stage-turn.ts +++ /dev/null @@ -1,34 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - - -/** - * An assistant response turn within a stage. - */ -export interface AssistantStageTurn { - 'kind': AssistantStageTurnKindEnum; - /** - * Assistant response text. - */ - 'content': string; -} - -export const AssistantStageTurnKindEnum = { - ASSISTANT: 'assistant' -} as const; - -export type AssistantStageTurnKindEnum = typeof AssistantStageTurnKindEnum[keyof typeof AssistantStageTurnKindEnum]; - - diff --git a/lib/packages/fabro-api-client/src/models/board-column-definition.ts b/lib/packages/fabro-api-client/src/models/board-column-definition.ts index 20deba1f5..314cbcde1 100644 --- a/lib/packages/fabro-api-client/src/models/board-column-definition.ts +++ b/lib/packages/fabro-api-client/src/models/board-column-definition.ts @@ -21,3 +21,6 @@ export interface BoardColumnDefinition { 'id': BoardColumn; 'name': string; } + + + diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index 0b72c59bd..fcb9e05ef 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -10,7 +10,6 @@ export * from './artifact-batch-upload-manifest'; export * from './artifact-entry'; export * from './artifact-list-response'; export * from './artifacts-settings'; -export * from './assistant-stage-turn'; export * from './auth-method'; export * from './billed-token-counts'; export * from './billing-by-model'; @@ -147,7 +146,6 @@ export * from './paginated-run-file-list'; export * from './paginated-run-list'; export * from './paginated-run-stage-list'; export * from './paginated-saved-query-list'; -export * from './paginated-stage-turn-list'; export * from './pagination-meta'; export * from './pending-interview-record'; export * from './pre-run-push-outcome'; @@ -279,7 +277,6 @@ export * from './stage-completion'; export * from './stage-outcome'; export * from './stage-projection'; export * from './stage-state'; -export * from './stage-turn'; export * from './start-run-request'; export * from './submit-answer-request'; export * from './success-reason'; @@ -287,13 +284,10 @@ export * from './system-actor-kind'; export * from './system-features'; export * from './system-info-response'; export * from './system-run-counts'; -export * from './system-stage-turn'; export * from './teams-integration-settings'; export * from './terminal-status'; export * from './timeline-entry-response'; export * from './tls-mode'; -export * from './tool-stage-turn'; -export * from './tool-use'; export * from './user-response'; export * from './validate-response'; export * from './webhook-strategy'; diff --git a/lib/packages/fabro-api-client/src/models/manifest-git.ts b/lib/packages/fabro-api-client/src/models/manifest-git.ts deleted file mode 100644 index bef882689..000000000 --- a/lib/packages/fabro-api-client/src/models/manifest-git.ts +++ /dev/null @@ -1,41 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - -// May contain unused imports in some cases -// @ts-ignore -import type { ManifestPreRunPushOutcome } from './manifest-pre-run-push-outcome'; - -/** - * Observable git state from the CLI working directory. - */ -export interface ManifestGit { - /** - * Remote origin URL with any embedded credentials removed. - */ - 'origin_url': string; - /** - * Current branch name. - */ - 'branch': string; - /** - * Current commit SHA. - */ - 'sha': string; - /** - * Whether the working tree has uncommitted changes. - */ - 'clean': boolean; - 'push_outcome': ManifestPreRunPushOutcome; -} diff --git a/lib/packages/fabro-api-client/src/models/manifest-pre-run-push-outcome.ts b/lib/packages/fabro-api-client/src/models/manifest-pre-run-push-outcome.ts deleted file mode 100644 index 93f6be234..000000000 --- a/lib/packages/fabro-api-client/src/models/manifest-pre-run-push-outcome.ts +++ /dev/null @@ -1,36 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - - -/** - * Outcome of the CLI\'s best-effort pre-run push. - */ -export interface ManifestPreRunPushOutcome { - 'type': ManifestPreRunPushOutcomeTypeEnum; - 'remote'?: string | null; - 'branch'?: string | null; - 'message'?: string | null; - 'repo_origin_url'?: string | null; -} - -export const ManifestPreRunPushOutcomeTypeEnum = { - NOT_ATTEMPTED: 'not_attempted', - SUCCEEDED: 'succeeded', - FAILED: 'failed', - SKIPPED_NO_REMOTE: 'skipped_no_remote', - SKIPPED_REMOTE_MISMATCH: 'skipped_remote_mismatch' -} as const; - -export type ManifestPreRunPushOutcomeTypeEnum = typeof ManifestPreRunPushOutcomeTypeEnum[keyof typeof ManifestPreRunPushOutcomeTypeEnum]; diff --git a/lib/packages/fabro-api-client/src/models/paginated-stage-turn-list.ts b/lib/packages/fabro-api-client/src/models/paginated-stage-turn-list.ts deleted file mode 100644 index 26f8d9cf2..000000000 --- a/lib/packages/fabro-api-client/src/models/paginated-stage-turn-list.ts +++ /dev/null @@ -1,30 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - -// May contain unused imports in some cases -// @ts-ignore -import type { PaginationMeta } from './pagination-meta'; -// May contain unused imports in some cases -// @ts-ignore -import type { StageTurn } from './stage-turn'; - -/** - * Paginated list of stage turns. - */ -export interface PaginatedStageTurnList { - 'data': Array; - 'meta': PaginationMeta; -} - diff --git a/lib/packages/fabro-api-client/src/models/pre-run-git-context.ts b/lib/packages/fabro-api-client/src/models/pre-run-git-context.ts deleted file mode 100644 index aaddc93e7..000000000 --- a/lib/packages/fabro-api-client/src/models/pre-run-git-context.ts +++ /dev/null @@ -1,30 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - -// May contain unused imports in some cases -// @ts-ignore -import type { DirtyStatus } from './dirty-status'; -// May contain unused imports in some cases -// @ts-ignore -import type { PreRunPushOutcome } from './pre-run-push-outcome'; - -/** - * Submitter-side git context captured before run creation. - */ -export interface PreRunGitContext { - 'display_base_sha'?: string | null; - 'local_dirty': DirtyStatus; - 'push_outcome': PreRunPushOutcome; -} diff --git a/lib/packages/fabro-api-client/src/models/set-secret-request.ts b/lib/packages/fabro-api-client/src/models/set-secret-request.ts deleted file mode 100644 index b343832c1..000000000 --- a/lib/packages/fabro-api-client/src/models/set-secret-request.ts +++ /dev/null @@ -1,25 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - - -/** - * Request to store a secret value. - */ -export interface SetSecretRequest { - /** - * The secret value to store. - */ - 'value': string; -} diff --git a/lib/packages/fabro-api-client/src/models/stage-turn.ts b/lib/packages/fabro-api-client/src/models/stage-turn.ts deleted file mode 100644 index 3c5ffdf3d..000000000 --- a/lib/packages/fabro-api-client/src/models/stage-turn.ts +++ /dev/null @@ -1,35 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - -// May contain unused imports in some cases -// @ts-ignore -import type { AssistantStageTurn } from './assistant-stage-turn'; -// May contain unused imports in some cases -// @ts-ignore -import type { SystemStageTurn } from './system-stage-turn'; -// May contain unused imports in some cases -// @ts-ignore -import type { ToolStageTurn } from './tool-stage-turn'; -// May contain unused imports in some cases -// @ts-ignore -import type { ToolUse } from './tool-use'; - -/** - * @type StageTurn - * A single turn in a stage conversation — a system prompt, assistant response, or tool invocation block. - */ -export type StageTurn = { kind: 'assistant' } & AssistantStageTurn | { kind: 'system' } & SystemStageTurn | { kind: 'tool' } & ToolStageTurn; - - diff --git a/lib/packages/fabro-api-client/src/models/system-stage-turn.ts b/lib/packages/fabro-api-client/src/models/system-stage-turn.ts deleted file mode 100644 index 68ad6386f..000000000 --- a/lib/packages/fabro-api-client/src/models/system-stage-turn.ts +++ /dev/null @@ -1,34 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - - -/** - * A system prompt turn that sets the stage\'s instructions. - */ -export interface SystemStageTurn { - 'kind': SystemStageTurnKindEnum; - /** - * System prompt text. - */ - 'content': string; -} - -export const SystemStageTurnKindEnum = { - SYSTEM: 'system' -} as const; - -export type SystemStageTurnKindEnum = typeof SystemStageTurnKindEnum[keyof typeof SystemStageTurnKindEnum]; - - diff --git a/lib/packages/fabro-api-client/src/models/tool-stage-turn.ts b/lib/packages/fabro-api-client/src/models/tool-stage-turn.ts deleted file mode 100644 index e637a0520..000000000 --- a/lib/packages/fabro-api-client/src/models/tool-stage-turn.ts +++ /dev/null @@ -1,41 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - -// May contain unused imports in some cases -// @ts-ignore -import type { ToolUse } from './tool-use'; - -/** - * A tool invocation turn containing one or more tool calls. - */ -export interface ToolStageTurn { - 'kind': ToolStageTurnKindEnum; - /** - * Text accompanying the tool invocations, or null when the turn contains only tool calls. - */ - 'content'?: string; - /** - * Tool invocations executed in this turn. - */ - 'tools': Array; -} - -export const ToolStageTurnKindEnum = { - TOOL: 'tool' -} as const; - -export type ToolStageTurnKindEnum = typeof ToolStageTurnKindEnum[keyof typeof ToolStageTurnKindEnum]; - - diff --git a/lib/packages/fabro-api-client/src/models/tool-use.ts b/lib/packages/fabro-api-client/src/models/tool-use.ts deleted file mode 100644 index c6f5e721e..000000000 --- a/lib/packages/fabro-api-client/src/models/tool-use.ts +++ /dev/null @@ -1,46 +0,0 @@ -/* tslint:disable */ -/* eslint-disable */ -/** - * Fabro Run API - * HTTP API for managing Fabro workflow run executions. - * - * The version of the OpenAPI document: 0.1.0 - * - * - * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech). - * https://openapi-generator.tech - * Do not edit the class manually. - */ - - - -/** - * A single tool invocation with its input, result, and execution metadata. - */ -export interface ToolUse { - /** - * Unique identifier for this tool invocation. Enables correlation in parallel tool use. - */ - 'id': string; - /** - * Name of the tool that was invoked. - */ - 'tool_name': string; - /** - * JSON-encoded input passed to the tool. - */ - 'input': string; - /** - * Output returned by the tool. Contains the error message when is_error is true. - */ - 'result': string; - /** - * Whether the tool invocation failed. When true, the result field contains the error message. - */ - 'is_error': boolean; - /** - * Wall-clock execution time of the tool invocation in milliseconds. - */ - 'duration_ms'?: number; -} - From e901cd3a813dc82f16738ab6bba9154b55ae1fea Mon Sep 17 00:00:00 2001 From: "fabro-sh-0530[bot]" <281434857+fabro-sh-0530[bot]@users.noreply.github.com> Date: Tue, 5 May 2026 09:18:05 -0400 Subject: [PATCH 2/2] Surface silent fallback warnings in runs and logs (#205) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Summary Silent fallback paths now emit stable warnings instead of degrading without a user-visible signal. The fallback behavior is unchanged; runs still continue, but worktree, Git, checkpoint, and LLM failover issues now show up in the run feed and logs. ### Plan Summary - Emit run notices for workflow Git/worktree fallback paths. - Reuse the existing failover event for one-shot LLM provider fallback. - Add tracing for sandbox pipe drain failures. ### What changed - Added `worktree_skipped_no_git` and gated `sandbox_git_unavailable` notices during initialization. - Added `git_push_failed` and `parallel_base_checkpoint_failed` notices, including redacted output tails where available. - Logged GitHub token mint failures with a structured `error` field before the existing notice. - Plumbed `Emitter` and `StageScope` through `CodergenBackend::one_shot` so the API backend emits the existing `agent.failover` event instead of a duplicate tracing-only warning. - Extracted sandbox pipe draining into a helper that warns on stdout/stderr read failures, with unit coverage for the error path. - Updated CLI snapshots for the new worktree warning in stderr and JSON event output. ⚒️ Generated with [Fabro](https://fabro.sh) --------- Co-authored-by: Fabro Co-authored-by: Bryan Helmkamp --- lib/crates/fabro-cli/src/commands/run/logs.rs | 51 +++++++++++++--- .../src/commands/run/run_progress/event.rs | 7 ++- .../src/commands/run/run_progress/mod.rs | 14 +++-- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 17 ++++++ lib/crates/fabro-cli/tests/it/cmd/dump.rs | 6 +- lib/crates/fabro-cli/tests/it/cmd/run.rs | 1 + .../tests/it/workflow/dry_run_examples.rs | 5 ++ lib/crates/fabro-sandbox/src/local.rs | 59 ++++++++++++++----- lib/crates/fabro-types/src/lib.rs | 2 +- lib/crates/fabro-types/src/run_event/infra.rs | 53 +++++++++++++---- lib/crates/fabro-types/src/run_event/mod.rs | 4 +- lib/crates/fabro-workflow/src/event.rs | 2 +- .../fabro-workflow/src/event/convert.rs | 6 +- .../fabro-workflow/src/event/emitter.rs | 15 ++--- .../fabro-workflow/src/handler/agent.rs | 2 + .../fabro-workflow/src/handler/llm/api.rs | 20 ++++--- .../fabro-workflow/src/handler/llm/cli.rs | 6 +- .../fabro-workflow/src/handler/parallel.rs | 8 ++- .../fabro-workflow/src/handler/prompt.rs | 23 ++++++-- .../fabro-workflow/src/lifecycle/artifact.rs | 10 ++-- .../fabro-workflow/src/lifecycle/git.rs | 43 +++++++++----- .../fabro-workflow/src/pipeline/finalize.rs | 42 +++++++++---- .../fabro-workflow/src/pipeline/initialize.rs | 59 +++++++++++++++---- .../src/pipeline/pull_request.rs | 4 +- .../fabro-workflow/tests/it/integration.rs | 2 + 25 files changed, 337 insertions(+), 124 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 8f01378cc..7d7e9063b 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -14,7 +14,7 @@ use std::time::Duration; use anyhow::{Context, Result, bail}; use chrono::{DateTime, Utc}; use fabro_redact::redact_jsonl_line; -use fabro_types::run_event::is_metadata_snapshot_compat_notice_code; +use fabro_types::RunNoticeCode; use fabro_util::json::normalize_json_value; use fabro_util::terminal::Styles; use tokio::time; @@ -801,7 +801,9 @@ fn format_event_pretty_value(envelope: &serde_json::Value, styles: &Styles) -> O } fn is_metadata_snapshot_compat_notice(envelope: &serde_json::Value) -> bool { - prop_str_field(envelope, "code").is_some_and(is_metadata_snapshot_compat_notice_code) + prop_str_field(envelope, "code") + .and_then(|code| code.parse::().ok()) + .is_some_and(RunNoticeCode::is_metadata_snapshot_compat) } fn str_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> { @@ -1150,14 +1152,27 @@ mod tests { #[test] fn pretty_run_notice_warn() { let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"run.notice","properties":{"level":"warn","code":"sandbox_cleanup_failed","message":"sandbox cleanup failed: boom"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); + let code = RunNoticeCode::SandboxCleanupFailed.to_string(); + let line = serde_json::json!({ + "ts": "2026-01-01T14:25:00Z", + "event": "run.notice", + "properties": { + "level": "warn", + "code": code, + "message": "sandbox cleanup failed: boom", + }, + }) + .to_string(); + let result = format_event_pretty(&line, &styles).unwrap(); assert!(result.contains("Warning:"), "got: {result}"); assert!( result.contains("sandbox cleanup failed: boom"), "got: {result}" ); - assert!(result.contains("[sandbox_cleanup_failed]"), "got: {result}"); + assert!( + result.contains(&format!("[{}]", RunNoticeCode::SandboxCleanupFailed)), + "got: {result}" + ); } #[test] @@ -1235,13 +1250,31 @@ mod tests { fn pretty_stream_suppresses_metadata_compat_notice_only() { let styles = no_color_styles(); let failed = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.failed","properties":{"phase":"checkpoint","branch":"fabro/meta","duration_ms":900,"failure_kind":"write","error":"write failed"}}"#; - let compat_notice = r#"{"ts":"2026-01-01T14:25:01Z","event":"run.notice","properties":{"level":"warn","code":"checkpoint_metadata_write_failed","message":"legacy metadata warning"}}"#; - let degraded_notice = r#"{"ts":"2026-01-01T14:25:02Z","event":"run.notice","properties":{"level":"warn","code":"checkpoint_metadata_degraded","message":"metadata snapshots disabled"}}"#; + let compat_notice = serde_json::json!({ + "ts": "2026-01-01T14:25:01Z", + "event": "run.notice", + "properties": { + "level": "warn", + "code": RunNoticeCode::CheckpointMetadataWriteFailed, + "message": "legacy metadata warning", + }, + }) + .to_string(); + let degraded_notice = serde_json::json!({ + "ts": "2026-01-01T14:25:02Z", + "event": "run.notice", + "properties": { + "level": "warn", + "code": RunNoticeCode::CheckpointMetadataDegraded, + "message": "metadata snapshots disabled", + }, + }) + .to_string(); let mut state = PrettyEventState::default(); assert!(format_event_pretty_streamed(failed, &styles, &mut state).is_some()); - assert!(format_event_pretty_streamed(compat_notice, &styles, &mut state).is_none()); - let degraded = format_event_pretty_streamed(degraded_notice, &styles, &mut state).unwrap(); + assert!(format_event_pretty_streamed(&compat_notice, &styles, &mut state).is_none()); + let degraded = format_event_pretty_streamed(°raded_notice, &styles, &mut state).unwrap(); assert!( degraded.contains("metadata snapshots disabled"), "got: {degraded}" diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs index 9e40f0d8b..ec56dce46 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/event.rs @@ -527,7 +527,7 @@ fn display_value(value: &Value) -> Option { mod tests { use fabro_agent::AgentEvent; use fabro_types::{MetadataSnapshotFailureKind, MetadataSnapshotPhase, fixtures}; - use fabro_workflow::event::{Event, to_run_event}; + use fabro_workflow::event::{Event, RunNoticeCode, to_run_event}; use super::*; @@ -804,10 +804,11 @@ mod tests { fn round_trip_run_notice() { let event = Event::RunNotice { level: RunNoticeLevel::Warn, - code: "sandbox_cleanup_failed".into(), + code: RunNoticeCode::SandboxCleanupFailed.to_string(), message: "sandbox cleanup failed".into(), exec_output_tail: None, }; + let expected_code = RunNoticeCode::SandboxCleanupFailed.to_string(); let stored = to_run_event(&fixtures::RUN_1, &event); let parsed = from_run_event(&stored).unwrap(); @@ -817,7 +818,7 @@ mod tests { level: RunNoticeLevel::Warn, code, message, - } if code == "sandbox_cleanup_failed" && message == "sandbox cleanup failed" + } if code == expected_code && message == "sandbox cleanup failed" )); } diff --git a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs index 2004ac0ba..7e7df7f2d 100644 --- a/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs @@ -3,8 +3,7 @@ reason = "sync CLI run-progress renderer: writes to std::io::stderr directly" )] -use fabro_types::RunEvent; -use fabro_types::run_event::is_metadata_snapshot_compat_notice_code; +use fabro_types::{RunEvent, RunNoticeCode}; mod event; mod info_display; @@ -444,7 +443,10 @@ impl ProgressUI { message, } => { if self.saw_metadata_snapshot_failure - && is_metadata_snapshot_compat_notice_code(&code) + && code + .parse::() + .ok() + .is_some_and(RunNoticeCode::is_metadata_snapshot_compat) { return; } @@ -1208,7 +1210,7 @@ mod tests { emit(&mut ui, Event::RunNotice { level: RunNoticeLevel::Warn, - code: "sandbox_cleanup_failed".into(), + code: RunNoticeCode::SandboxCleanupFailed.to_string(), message: "sandbox cleanup failed".into(), exec_output_tail: None, }); @@ -1279,13 +1281,13 @@ mod tests { }); emit(&mut ui, Event::RunNotice { level: RunNoticeLevel::Warn, - code: "checkpoint_metadata_write_failed".into(), + code: RunNoticeCode::CheckpointMetadataWriteFailed.to_string(), message: "legacy metadata warning".into(), exec_output_tail: None, }); emit(&mut ui, Event::RunNotice { level: RunNoticeLevel::Warn, - code: "checkpoint_metadata_degraded".into(), + code: RunNoticeCode::CheckpointMetadataDegraded.to_string(), message: "metadata snapshots are disabled for this run".into(), exec_output_tail: None, }); diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index e703e3207..7c6728f39 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -160,6 +160,7 @@ fn attach_replays_completed_detached_run() { ----- stdout ----- ----- stderr ----- Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ Start [TIME] ✓ Run Tests [TIME] @@ -267,6 +268,7 @@ fn attach_before_completion_streams_to_finished_state() { ----- stdout ----- ----- stderr ----- Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ start [DURATION] ✓ wait [DURATION] @@ -699,6 +701,21 @@ fn attach_json_errors_without_prompting_for_human_input() { "run_id": "[ULID]", "ts": "[TIMESTAMP]" }, + { + "actor": { + "kind": "worker", + "run_id": "[ULID]" + }, + "event": "run.notice", + "id": "[EVENT_ID]", + "properties": { + "code": "worktree_skipped_no_git", + "level": "warn", + "message": "Worktree mode `always` requested but no Git repository was found; running without a worktree." + }, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" + }, { "actor": { "kind": "worker", diff --git a/lib/crates/fabro-cli/tests/it/cmd/dump.rs b/lib/crates/fabro-cli/tests/it/cmd/dump.rs index 12b13ee9a..c541b59eb 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/dump.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/dump.rs @@ -278,9 +278,9 @@ fn dump_exports_completed_run_snapshot() { "); assert_snapshot!(dump_file_summary(&output_dir), @" - checkpoints/0013.json - checkpoints/0017.json - checkpoints/0021.json + checkpoints/0014.json + checkpoints/0018.json + checkpoints/0022.json events.jsonl graph.fabro run.json diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 19d876cfc..2aa614979 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -684,6 +684,7 @@ fn dry_run_simple() { Run: [ULID] Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ Start [TIME] ✓ Run Tests [TIME] diff --git a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs index bb8a50c85..c3f8d836c 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs @@ -21,6 +21,7 @@ fn dry_run_branching() { warning [node: implement]: Node 'implement' has goal_gate=true but no retry_target or fallback_retry_target (goal_gate_has_retry) Run: [ULID] Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ Start [TIME] ✓ Plan [TIME] @@ -57,6 +58,7 @@ fn dry_run_conditions() { Run: [ULID] Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ start [TIME] ✓ Decide [TIME] @@ -91,6 +93,7 @@ fn dry_run_parallel() { Run: [ULID] Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ start [TIME] ✓ Fork Work [TIME] @@ -126,6 +129,7 @@ fn dry_run_styled() { Run: [ULID] Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ start [TIME] ✓ Plan [TIME] @@ -161,6 +165,7 @@ fn dry_run_legacy_tool() { Run: [ULID] Web UI: http://localhost:3000/runs/[ULID] + Warning: Worktree mode `always` requested but no Git repository was found; running without a worktree. [worktree_skipped_no_git] Sandbox: local (ready in [TIME]) ✓ Start [TIME] ✓ Echo [TIME] diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index 8b69b7377..66d084b5f 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/lib/crates/fabro-sandbox/src/local.rs @@ -126,6 +126,19 @@ fn process_env_vars() -> Vec<(String, String)> { std::env::vars().collect() } +async fn drain_pipe(mut pipe: Option, stream: CommandOutputStream) -> String +where + R: AsyncRead + Unpin, +{ + let mut buf = String::new(); + if let Some(ref mut reader) = pipe { + if let Err(err) = reader.read_to_string(&mut buf).await { + tracing::warn!(error = %err, ?stream, "Failed to drain child output"); + } + } + buf +} + #[async_trait] impl Sandbox for LocalSandbox { async fn read_file( @@ -277,22 +290,12 @@ impl Sandbox for LocalSandbox { // it writes more than the OS pipe buffer (~64 KB) the write() syscall // blocks until the parent drains the pipe, but the parent is blocked // on child.wait(). - let mut stdout_pipe = child.stdout.take(); - let mut stderr_pipe = child.stderr.take(); - let stdout_task = tokio::spawn(async move { - let mut buf = String::new(); - if let Some(ref mut r) = stdout_pipe { - let _ = r.read_to_string(&mut buf).await; - } - buf - }); - let stderr_task = tokio::spawn(async move { - let mut buf = String::new(); - if let Some(ref mut r) = stderr_pipe { - let _ = r.read_to_string(&mut buf).await; - } - buf - }); + let stdout_pipe = child.stdout.take(); + let stderr_pipe = child.stderr.take(); + let stdout_task = + tokio::spawn(async move { drain_pipe(stdout_pipe, CommandOutputStream::Stdout).await }); + let stderr_task = + tokio::spawn(async move { drain_pipe(stderr_pipe, CommandOutputStream::Stderr).await }); let (termination, exit_code) = tokio::select! { status_result = child.wait() => { @@ -712,7 +715,12 @@ where )] mod tests { use std::collections::HashMap; + use std::io; use std::path::PathBuf; + use std::pin::Pin; + use std::task::{Context as TaskContext, Poll}; + + use tokio::io::ReadBuf; use super::*; @@ -722,6 +730,25 @@ mod tests { dir } + #[tokio::test] + async fn drain_pipe_returns_empty_buffer_after_read_failure() { + struct FailingReader; + + impl AsyncRead for FailingReader { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut TaskContext<'_>, + _buf: &mut ReadBuf<'_>, + ) -> Poll> { + Poll::Ready(Err(io::Error::other("simulated read failure"))) + } + } + + let output = drain_pipe(Some(FailingReader), CommandOutputStream::Stdout).await; + + assert!(output.is_empty()); + } + #[tokio::test] async fn read_file_with_line_numbers() { let dir = temp_dir(); diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 22675855c..e5d3b65a5 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -72,7 +72,7 @@ pub use run::{ pub use run_blob_id::RunBlobId; pub use run_event::{ EventBody, ExecOutputTail, InterviewOption, MetadataSnapshotFailureKind, MetadataSnapshotPhase, - RunEvent, RunNoticeLevel, + RunEvent, RunNoticeCode, RunNoticeLevel, }; pub use run_id::{RunId, fixtures}; pub use run_projection::{PendingInterviewRecord, RunProjection, StageProjection, first_event_seq}; diff --git a/lib/crates/fabro-types/src/run_event/infra.rs b/lib/crates/fabro-types/src/run_event/infra.rs index 8ee1bc32c..77689087b 100644 --- a/lib/crates/fabro-types/src/run_event/infra.rs +++ b/lib/crates/fabro-types/src/run_event/infra.rs @@ -1,17 +1,48 @@ use serde::{Deserialize, Serialize}; -/// Legacy `run.notice` codes paired with the new `metadata.snapshot.failed` -/// event for backward compatibility. Display layers suppress these so the -/// typed event renders without a duplicate raw warning. -pub const NOTICE_CODE_CHECKPOINT_METADATA_WRITE_FAILED: &str = "checkpoint_metadata_write_failed"; -pub const NOTICE_CODE_CHECKPOINT_METADATA_PUSH_FAILED: &str = "checkpoint_metadata_push_failed"; +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + Hash, + Serialize, + Deserialize, + strum::Display, + strum::EnumString, + strum::IntoStaticStr, +)] +#[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] +pub enum RunNoticeCode { + ArtifactCollectionFailed, + ArtifactOffloadFailed, + ArtifactSyncFailed, + ArtifactUploadFailed, + CheckpointMetadataDegraded, + CheckpointMetadataPushFailed, + CheckpointMetadataWriteFailed, + DirtyWorktree, + GitDiffFailed, + GitPushFailed, + GithubTokenFailed, + ParallelBaseCheckpointFailed, + PullRequestFailed, + SandboxCleanupFailed, + SandboxGitUnavailable, + SandboxPreserved, + WorktreeSkippedNoGit, +} -#[must_use] -pub fn is_metadata_snapshot_compat_notice_code(code: &str) -> bool { - matches!( - code, - NOTICE_CODE_CHECKPOINT_METADATA_WRITE_FAILED | NOTICE_CODE_CHECKPOINT_METADATA_PUSH_FAILED - ) +impl RunNoticeCode { + #[must_use] + pub fn is_metadata_snapshot_compat(self) -> bool { + matches!( + self, + Self::CheckpointMetadataWriteFailed | Self::CheckpointMetadataPushFailed + ) + } } #[derive( diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index 9d5fbf11d..647efc381 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -1366,7 +1366,7 @@ mod tests { for body in [ EventBody::RunNotice(RunNoticeProps { level: RunNoticeLevel::Warn, - code: "git_diff_failed".to_string(), + code: RunNoticeCode::GitDiffFailed.to_string(), message: "git diff failed".to_string(), exec_output_tail: Some(tail.clone()), }), @@ -1402,7 +1402,7 @@ mod tests { for body in [ EventBody::RunNotice(RunNoticeProps { level: RunNoticeLevel::Warn, - code: "git_diff_failed".to_string(), + code: RunNoticeCode::GitDiffFailed.to_string(), message: "git diff failed".to_string(), exec_output_tail: None, }), diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index f5120e62e..e64b2ae30 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -8,7 +8,7 @@ mod stored_fields; #[cfg(test)] mod test_support; -pub use fabro_types::{EventBody, RunNoticeLevel}; +pub use fabro_types::{EventBody, RunNoticeCode, RunNoticeLevel}; pub use self::convert::{to_run_event, to_run_event_at}; pub use self::emitter::Emitter; diff --git a/lib/crates/fabro-workflow/src/event/convert.rs b/lib/crates/fabro-workflow/src/event/convert.rs index 4aa9df5b1..255d235ea 100644 --- a/lib/crates/fabro-workflow/src/event/convert.rs +++ b/lib/crates/fabro-workflow/src/event/convert.rs @@ -1169,8 +1169,8 @@ mod tests { use std::collections::BTreeMap; use ::fabro_types::{ - EventBody, FailureReason, ParallelBranchId, Principal, RunNoticeLevel, RunProvenance, - StageId, SystemActorKind, fixtures, run_event as fabro_types, + EventBody, FailureReason, ParallelBranchId, Principal, RunNoticeCode, RunNoticeLevel, + RunProvenance, StageId, SystemActorKind, fixtures, run_event as fabro_types, }; use chrono::Utc; use fabro_agent::{AgentEvent, SandboxEvent}; @@ -1643,7 +1643,7 @@ mod tests { fn run_notice_maps_exec_output_tail_to_props() { let stored = to_run_event(&fixtures::RUN_1, &Event::RunNotice { level: RunNoticeLevel::Warn, - code: "git_diff_failed".to_string(), + code: RunNoticeCode::GitDiffFailed.to_string(), message: "git diff failed".to_string(), exec_output_tail: Some(exec_tail()), }); diff --git a/lib/crates/fabro-workflow/src/event/emitter.rs b/lib/crates/fabro-workflow/src/event/emitter.rs index a1bba727a..dc50a56dd 100644 --- a/lib/crates/fabro-workflow/src/event/emitter.rs +++ b/lib/crates/fabro-workflow/src/event/emitter.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicI64, Ordering}; -use ::fabro_types::{ExecOutputTail, RunEvent, RunId, RunNoticeLevel}; +use ::fabro_types::{ExecOutputTail, RunEvent, RunId, RunNoticeCode, RunNoticeLevel}; use chrono::Utc; use fabro_agent::{WorktreeEvent, WorktreeEventCallback}; @@ -76,15 +76,10 @@ impl Emitter { self.emit_with_scope(event, Some(scope)); } - pub fn notice( - &self, - level: RunNoticeLevel, - code: impl Into, - message: impl Into, - ) { + pub fn notice(&self, level: RunNoticeLevel, code: RunNoticeCode, message: impl Into) { self.emit(&Event::RunNotice { level, - code: code.into(), + code: code.to_string(), message: message.into(), exec_output_tail: None, }); @@ -93,13 +88,13 @@ impl Emitter { pub fn notice_with_tail( &self, level: RunNoticeLevel, - code: impl Into, + code: RunNoticeCode, message: impl Into, exec_output_tail: Option, ) { self.emit(&Event::RunNotice { level, - code: code.into(), + code: code.to_string(), message: message.into(), exec_output_tail, }); diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 676e2f142..7e3a4be0c 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -52,6 +52,8 @@ pub trait CodergenBackend: Send + Sync { _node: &Node, _prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { Err(Error::Validation( "one_shot mode not supported by this backend".into(), diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index 5d21c7d32..9247c19ec 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -285,6 +285,8 @@ impl CodergenBackend for AgentApiBackend { node: &Node, prompt: &str, system_prompt: Option<&str>, + emitter: &Arc, + stage_scope: &StageScope, ) -> Result { let client = Client::from_source(self.source.as_ref()) .await @@ -358,14 +360,16 @@ impl CodergenBackend for AgentApiBackend { let mut found = None; for target in fallback_chain { - tracing::warn!( - stage = node.id.as_str(), - from_provider = from_provider.as_str(), - from_model = from_model.as_str(), - to_provider = target.provider.as_str(), - to_model = target.model.as_str(), - error = error_msg.as_str(), - "LLM provider failover (prompt)" + emitter.emit_scoped( + &Event::Failover { + stage: node.id.clone(), + from_provider: from_provider.clone(), + from_model: from_model.clone(), + to_provider: target.provider.clone(), + to_model: target.model.clone(), + error: error_msg.clone(), + }, + stage_scope, ); let max_tokens = node.max_tokens().or_else(|| { diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 5036eec97..7fcf60ed8 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -810,9 +810,13 @@ impl CodergenBackend for BackendRouter { node: &Node, prompt: &str, system_prompt: Option<&str>, + emitter: &Arc, + stage_scope: &StageScope, ) -> Result { // CLI backend doesn't support one_shot, always route to API - self.api_backend.one_shot(node, prompt, system_prompt).await + self.api_backend + .one_shot(node, prompt, system_prompt, emitter, stage_scope) + .await } } diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 39862586c..c58104f83 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -12,7 +12,7 @@ use tokio::sync::Semaphore; use super::{EngineServices, Handler}; use crate::context::{Context, WorkflowContext, keys}; use crate::error::Error; -use crate::event::{Event, StageScope}; +use crate::event::{Event, RunNoticeCode, RunNoticeLevel, StageScope}; use crate::git::sanitize_ref_component; use crate::hook_context::set_hook_node; use crate::millis_u64; @@ -207,6 +207,12 @@ impl Handler for ParallelHandler { error = %fabro_sandbox::display_for_log(&e), "parallel base checkpoint failed" ); + services.run.emitter.notice_with_tail( + RunNoticeLevel::Warn, + RunNoticeCode::ParallelBaseCheckpointFailed, + format!("Could not checkpoint base state before parallel branches: {e}"), + fabro_sandbox::default_redacted_output_tail(&e), + ); None } } diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index a76099b80..caad1e058 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -105,7 +105,13 @@ impl Handler for PromptHandler { let (response_text, stage_usage, backend_files_touched) = if let Some(backend) = &self.backend { let result = backend - .one_shot(node, &prompt, system_prompt.as_deref()) + .one_shot( + node, + &prompt, + system_prompt.as_deref(), + &services.run.emitter, + &stage_scope, + ) .await; match result { Ok(CodergenResult::Full(outcome)) => return Ok(outcome), @@ -187,6 +193,7 @@ mod tests { use tempfile::TempDir; use super::*; + use crate::event::Emitter; fn make_services() -> EngineServices { EngineServices::test_default() @@ -211,7 +218,7 @@ mod tests { let mut services = EngineServices::test_default(); services.run = services .run - .with_emitter(Arc::new(crate::event::Emitter::new(fixtures::RUN_1))) + .with_emitter(Arc::new(Emitter::new(fixtures::RUN_1))) .with_run_store(run_store.clone().into()); let logger = crate::event::StoreProgressLogger::new(run_store.clone()); logger.register(services.run.emitter.as_ref()); @@ -267,7 +274,7 @@ mod tests { _prompt: &str, _context: &Context, _thread_id: Option<&str>, - _emitter: &Arc, + _emitter: &Arc, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { @@ -279,6 +286,8 @@ mod tests { _node: &Node, _prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { Ok(CodergenResult::Text { text: "one-shot response".to_string(), @@ -327,7 +336,7 @@ mod tests { _prompt: &str, _context: &Context, _thread_id: Option<&str>, - _emitter: &Arc, + _emitter: &Arc, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { @@ -339,6 +348,8 @@ mod tests { _node: &Node, _prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { Ok(CodergenResult::Text { text: "one-shot response".to_string(), @@ -384,7 +395,7 @@ mod tests { _prompt: &str, _context: &Context, _thread_id: Option<&str>, - _emitter: &Arc, + _emitter: &Arc, _sandbox: &Arc, _tool_hooks: Option>, ) -> Result { @@ -396,6 +407,8 @@ mod tests { _node: &Node, prompt: &str, system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, ) -> Result { *self.captured_prompt.lock().unwrap() = Some(prompt.to_string()); *self.captured_system_prompt.lock().unwrap() = Some(system_prompt.map(String::from)); diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index f65fcee86..50d6ccded 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -16,7 +16,7 @@ use tokio::time::sleep; use crate::artifact::{normalize_durable_updates, offload_large_values, sync_artifacts_to_env}; use crate::artifact_snapshot::collect_artifacts; use crate::artifact_upload::ArtifactSink; -use crate::event::{Emitter, Event, RunNoticeLevel}; +use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel}; use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::lifecycle::event::{stage_scope_for, stage_visit}; use crate::outcome::BilledModelUsage; @@ -125,7 +125,7 @@ impl RunLifecycle for ArtifactLifecycle { { self.emitter.notice( RunNoticeLevel::Warn, - "artifact_upload_failed", + RunNoticeCode::ArtifactUploadFailed, format!("[node: {node_id}] artifact upload failed: {err}"), ); return Ok(()); @@ -151,7 +151,7 @@ impl RunLifecycle for ArtifactLifecycle { Err(e) => { self.emitter.notice( RunNoticeLevel::Warn, - "artifact_collection_failed", + RunNoticeCode::ArtifactCollectionFailed, format!("[node: {node_id}] artifact collection failed: {e}"), ); } @@ -174,7 +174,7 @@ impl RunLifecycle for ArtifactLifecycle { { self.emitter.notice( RunNoticeLevel::Warn, - "artifact_offload_failed", + RunNoticeCode::ArtifactOffloadFailed, format!("[node: {node_id}] artifact offload failed: {e}"), ); } @@ -187,7 +187,7 @@ impl RunLifecycle for ArtifactLifecycle { { self.emitter.notice( RunNoticeLevel::Warn, - "artifact_sync_failed", + RunNoticeCode::ArtifactSyncFailed, format!("[node: {node_id}] artifact sync failed: {e}"), ); } diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 0e0e9f195..2f9d5bfdb 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -14,7 +14,7 @@ use fabro_util::error::collect_causes; use fabro_util::time::elapsed_ms; use crate::artifact; -use crate::event::{Emitter, Event, RunNoticeLevel, StageScope}; +use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::lifecycle::event::stage_scope_for; use crate::outcome::BilledModelUsage; @@ -128,7 +128,10 @@ impl RunLifecycle for GitLifecycle { None, None, ); - self.emit_metadata_warning("checkpoint_metadata_write_failed", message); + self.emit_metadata_warning( + RunNoticeCode::CheckpointMetadataWriteFailed, + message, + ); } }, Err(err) => { @@ -145,7 +148,10 @@ impl RunLifecycle for GitLifecycle { None, None, ); - self.emit_metadata_warning("checkpoint_metadata_write_failed", message); + self.emit_metadata_warning( + RunNoticeCode::CheckpointMetadataWriteFailed, + message, + ); } } } @@ -217,7 +223,7 @@ impl RunLifecycle for GitLifecycle { Some(&scope), ); self.emit_metadata_warning( - "checkpoint_metadata_write_failed", + RunNoticeCode::CheckpointMetadataWriteFailed, message, ); None @@ -239,7 +245,10 @@ impl RunLifecycle for GitLifecycle { None, Some(&scope), ); - self.emit_metadata_warning("checkpoint_metadata_write_failed", message); + self.emit_metadata_warning( + RunNoticeCode::CheckpointMetadataWriteFailed, + message, + ); None } } @@ -292,6 +301,12 @@ impl RunLifecycle for GitLifecycle { error = %fabro_sandbox::display_for_log(&err), "git push from run lifecycle failed" ); + self.emitter.notice_with_tail( + RunNoticeLevel::Warn, + RunNoticeCode::GitPushFailed, + format!("Failed to push run branch {branch}: {err}"), + exec_output_tail.clone(), + ); (false, exec_output_tail) } }; @@ -321,7 +336,7 @@ impl RunLifecycle for GitLifecycle { fabro_sandbox::default_redacted_output_tail(&err); self.emitter.notice_with_tail( RunNoticeLevel::Warn, - "git_diff_failed", + RunNoticeCode::GitDiffFailed, format!("[node: {node_id}] git diff failed: {err}"), exec_output_tail, ); @@ -395,7 +410,10 @@ impl GitLifecycle { Some(snapshot.bytes), scope, ); - self.emit_metadata_warning("checkpoint_metadata_push_failed", message); + self.emit_metadata_warning( + RunNoticeCode::CheckpointMetadataPushFailed, + message, + ); } else { self.emit_metadata_snapshot_completed( phase, @@ -421,7 +439,7 @@ impl GitLifecycle { None, scope, ); - self.emit_metadata_warning("checkpoint_metadata_write_failed", message); + self.emit_metadata_warning(RunNoticeCode::CheckpointMetadataWriteFailed, message); None } } @@ -506,14 +524,9 @@ impl GitLifecycle { } } - fn emit_metadata_warning(&self, code: &str, message: String) { + fn emit_metadata_warning(&self, code: RunNoticeCode, message: String) { if self.metadata_runtime.mark_metadata_degraded() { - self.emitter.emit(&Event::RunNotice { - level: RunNoticeLevel::Warn, - code: code.to_string(), - message, - exec_output_tail: None, - }); + self.emitter.notice(RunNoticeLevel::Warn, code, message); } } } diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 927d2d833..cb58bd345 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -10,7 +10,7 @@ use fabro_util::time::elapsed_ms; use super::types::{Concluded, FinalizeOptions, Retroed}; use crate::error::Error; -use crate::event::{Event, RunNoticeLevel}; +use crate::event::{Event, RunNoticeCode, RunNoticeLevel}; use crate::outcome::{Outcome, OutcomeExt, StageOutcome}; use crate::records::{Checkpoint, Conclusion, StageSummary}; use crate::run_metadata::MetadataSnapshot; @@ -239,7 +239,11 @@ pub async fn write_finalize_commit( None, None, ); - emit_metadata_warning(services, "checkpoint_metadata_write_failed", message); + emit_metadata_warning( + services, + RunNoticeCode::CheckpointMetadataWriteFailed, + message, + ); return; } }; @@ -260,7 +264,11 @@ pub async fn write_finalize_commit( None, None, ); - emit_metadata_warning(services, "checkpoint_metadata_write_failed", message); + emit_metadata_warning( + services, + RunNoticeCode::CheckpointMetadataWriteFailed, + message, + ); return; } }; @@ -281,7 +289,11 @@ pub async fn write_finalize_commit( Some(snapshot.entry_count), Some(snapshot.bytes), ); - emit_metadata_warning(services, "checkpoint_metadata_push_failed", message); + emit_metadata_warning( + services, + RunNoticeCode::CheckpointMetadataPushFailed, + message, + ); } else { emit_metadata_snapshot_completed(services, phase, meta_branch, started, &snapshot); } @@ -300,7 +312,11 @@ pub async fn write_finalize_commit( None, None, ); - emit_metadata_warning(services, "checkpoint_metadata_write_failed", message); + emit_metadata_warning( + services, + RunNoticeCode::CheckpointMetadataWriteFailed, + message, + ); } } } @@ -363,7 +379,7 @@ fn emit_metadata_snapshot_failed( }); } -fn emit_metadata_warning(services: &RunServices, code: &str, message: String) { +fn emit_metadata_warning(services: &RunServices, code: RunNoticeCode, message: String) { if services.metadata_runtime.mark_metadata_degraded() { services.emitter.notice(RunNoticeLevel::Warn, code, message); } @@ -387,7 +403,7 @@ async fn compute_final_patch( Err(err) => { services.emitter.notice( RunNoticeLevel::Warn, - "git_diff_failed", + RunNoticeCode::GitDiffFailed, format!("final diff failed: {err}"), ); None @@ -534,7 +550,7 @@ pub async fn finalize(retroed: Retroed, options: &FinalizeOptions) -> Result Result Result Option { if let Some(env_name) = env_name { options.emitter.notice( RunNoticeLevel::Warn, - "dirty_worktree", + RunNoticeCode::DirtyWorktree, format!("Uncommitted changes will not be included in the {env_name}."), ); } @@ -200,6 +200,14 @@ fn resolve_worktree_plan(options: &mut InitOptions) -> Option { }) } +fn worktree_skipped_notice(mode: Option) -> Option<(RunNoticeCode, &'static str)> { + matches!(mode, Some(WorktreeMode::Always)).then_some(( + RunNoticeCode::WorktreeSkippedNoGit, + "Worktree mode `always` requested but no Git repository was found; running without a \ + worktree.", + )) +} + fn git_setup_intent(run_options: &RunOptions) -> GitSetupIntent { if let Some(source) = run_options.fork_source_ref.as_ref() { GitSetupIntent::ForkFromCheckpoint { @@ -239,11 +247,14 @@ async fn build_sandbox_env( Ok(token) => { env.insert("GITHUB_TOKEN".to_string(), token); } - Err(e) => emitter.notice( - RunNoticeLevel::Warn, - "github_token_failed", - format!("Failed to mint GitHub token: {e}"), - ), + Err(e) => { + tracing::warn!(error = %e, "Failed to mint GitHub token"); + emitter.notice( + RunNoticeLevel::Warn, + RunNoticeCode::GithubTokenFailed, + format!("Failed to mint GitHub token: {e}"), + ); + } } } } @@ -516,6 +527,13 @@ pub async fn initialize( )) }; if worktree_plan.is_some() && !worktree_created { + if let Some((code, message)) = worktree_skipped_notice(options.worktree_mode) { + tracing::warn!( + worktree_mode = ?options.worktree_mode, + "worktree skipped: cwd is not a git repository" + ); + options.emitter.notice(RunNoticeLevel::Warn, code, message); + } options.run_options.git = None; } let cleanup_guard = scopeguard::guard(Arc::clone(&sandbox), |sandbox| { @@ -593,7 +611,8 @@ pub async fn initialize( .is_some(); if !has_run_branch { let intent = git_setup_intent(&options.run_options); - if sandbox.origin_url().is_some() { + let sandbox_has_origin = sandbox.origin_url().is_some(); + if sandbox_has_origin { sandbox_git .ensure_git_available(&*sandbox) .await @@ -619,7 +638,16 @@ pub async fn initialize( options.run_options.base_branch = info.base_branch; } } - Ok(None) => {} + Ok(None) => { + if sandbox_has_origin { + options.emitter.notice( + RunNoticeLevel::Warn, + RunNoticeCode::SandboxGitUnavailable, + "Sandbox could not set up Git despite a configured origin; running \ + without checkpointing or PR support.", + ); + } + } Err(e) => { return Err(Error::engine_with_source("Sandbox git setup failed", &e)); } @@ -702,7 +730,7 @@ pub async fn initialize( if metadata_runtime.mark_metadata_degraded() { options.emitter.notice( RunNoticeLevel::Warn, - "checkpoint_metadata_write_failed", + RunNoticeCode::CheckpointMetadataWriteFailed, message, ); } @@ -1011,6 +1039,17 @@ mod tests { assert!(options.run_options.git.is_none()); } + #[test] + fn worktree_skipped_notice_only_warns_for_always() { + assert!(worktree_skipped_notice(None).is_none()); + assert!(worktree_skipped_notice(Some(WorktreeMode::Clean)).is_none()); + assert!(worktree_skipped_notice(Some(WorktreeMode::Dirty)).is_none()); + assert!(worktree_skipped_notice(Some(WorktreeMode::Never)).is_none()); + + let (code, _) = worktree_skipped_notice(Some(WorktreeMode::Always)).unwrap(); + assert_eq!(code, RunNoticeCode::WorktreeSkippedNoGit); + } + #[tokio::test] async fn initialize_prepares_sandbox_and_uses_persisted_run_dir() { let temp = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 35f27aa92..37aec323e 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -14,7 +14,7 @@ use fabro_util::text::strip_goal_decoration; use tracing::{debug, info, warn}; use super::types::{Concluded, Finalized, PullRequestOptions}; -use crate::event::{Event, RunNoticeLevel}; +use crate::event::{Event, RunNoticeCode, RunNoticeLevel}; use crate::outcome::{StageOutcome, format_cost as outcome_format_cost}; use crate::records::{Conclusion, RunSpec}; use crate::runtime_store::RunStoreHandle; @@ -675,7 +675,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> .emit(&Event::PullRequestFailed { error: e.clone() }); services.emitter.notice( RunNoticeLevel::Warn, - "pull_request_failed", + RunNoticeCode::PullRequestFailed, format!("PR creation failed: {e}"), ); } diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index f6ece331b..b89fc7f5a 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -6221,6 +6221,8 @@ mod real_llm { _node: &Node, prompt: &str, _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &fabro_workflow::event::StageScope, ) -> Result { self.complete(prompt).await }