fabro(01KQT9NFG90GWYZ7CZ0FAH0E12): implement (succeeded)

Fabro-Run: 01KQT9NFG90GWYZ7CZ0FAH0E12
Fabro-Completed: 5
Fabro-Checkpoint: fea731b9a0

⚒️ Generated with [Fabro](https://fabro.sh)
This commit is contained in:
Fabro 2026-05-04 20:38:09 +00:00
parent 5776e861cd
commit 19bf07acc1
31 changed files with 961 additions and 637 deletions

View file

@ -217,6 +217,61 @@ export async function apiPaginatedFetcher<TItem, TExtra extends object = {}>(
}
}
function stageEventsPagePath(key: string, sinceSeq: number, limit: number): string {
const url = new URL(apiPath(key), "http://fabro.local");
url.searchParams.set("since_seq", String(sinceSeq));
url.searchParams.set("limit", String(limit));
return `${url.pathname}${url.search}`;
}
/**
* 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<TItem>(key: string): Promise<TItem[]> {
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<TItem & { seq: number }>;
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 lastSeq = page.data[page.data.length - 1].seq;
sinceSeq = lastSeq + 1;
}
}
export async function apiJsonMutation<TResponse, TArg = unknown>(
key: string,
{ arg }: { arg: TArg },
@ -233,4 +288,4 @@ export async function apiJsonMutation<TResponse, TArg = unknown>(
}
if (response.status === 204) return undefined as TResponse;
return response.json() as Promise<TResponse>;
}
}

View file

@ -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<PaginatedStageTurnList | null>(
id && stageId && enabled ? queryKeys.runs.stageTurns(id, stageId) : null,
apiNullableFetcher,
);
}
export function useRunEventsList(id: string | undefined, enabled = true) {
return useSWR<PaginatedEventList | null>(
id && enabled ? queryKeys.runs.events(id, 1000) : null,
apiNullableFetcher,
export function useRunStageEvents(id: string | undefined, stageId: string | undefined) {
return useSWR<EventEnvelope[]>(
id && stageId ? queryKeys.runs.stageEvents(id, stageId) : null,
fetchAllStageEvents<EventEnvelope>,
);
}
@ -205,4 +194,4 @@ export function useServerSettings() {
return useSWR<ServerSettings>(queryKeys.settings.server(), apiFetcher, immutableOptions);
}
export { apiTextFetcher };
export { apiTextFetcher };

View file

@ -23,7 +23,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([]);
});
});

View file

@ -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",
},
};
};

View file

@ -43,7 +43,19 @@ const RUN_SUMMARY_EVENTS = new Set([
"run.unarchived",
]);
const STAGE_EVENTS = new Set(["stage.started", "stage.completed", "stage.failed"]);
const COMMAND_EVENTS = new Set(["command.started", "command.completed"]);
// Every event type the `eventsToActivity` reducer in `routes/run-stages.tsx`
// consumes. When any of these arrive for a stage we currently view, the
// stage-events SWR key for that stage must be invalidated so the panel
// refetches. The lifecycle `STAGE_EVENTS` set is kept separate because it
// also fans out to run-scoped invalidations (stages list, graph, detail).
const STAGE_ACTIVITY_EVENTS = new Set([
"stage.prompt",
"agent.message",
"agent.tool.started",
"agent.tool.completed",
"command.started",
"command.completed",
]);
const INTERVIEW_EVENTS = new Set([
"interview.started",
"interview.completed",
@ -91,20 +103,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 [];
@ -178,4 +183,4 @@ export function useRunEvents(runId: string | undefined) {
if (!runId) return;
return subscribeToRunEvents(runId, mutate as MutateFn);
}, [mutate, runId]);
}
}

View file

@ -1,6 +1,7 @@
import { describe, expect, test } from "bun:test";
import type { EventEnvelope } from "@qltysh/fabro-api-client";
import { isSafeMarkdownHref } from "./run-stages";
import { eventsToActivity, isSafeMarkdownHref } from "./run-stages";
describe("isSafeMarkdownHref", () => {
test("rejects protocol-relative URLs", () => {
@ -15,3 +16,105 @@ describe("isSafeMarkdownHref", () => {
expect(isSafeMarkdownHref("mailto:test@example.com")).toBe(true);
});
});
function envelope(seq: number, partial: Partial<EventEnvelope>): EventEnvelope {
return {
seq,
id: `evt-${seq}`,
ts: "2026-04-09T12:00:00Z",
run_id: "run-1",
...partial,
} as EventEnvelope;
}
describe("eventsToActivity", () => {
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",
script: "cargo fmt",
language: "shell",
stdout: "ok",
exitCode: 0,
running: 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("filters out events for other node_ids", () => {
const events: EventEnvelope[] = [
envelope(1, {
event: "agent.message",
node_id: "other-stage",
properties: { text: "noise" },
}),
envelope(2, {
event: "agent.message",
node_id: "detect-drift",
properties: { text: "signal" },
}),
];
const turns = eventsToActivity(events, "detect-drift");
expect(turns).toHaveLength(1);
if (turns[0].kind === "assistant") {
expect(turns[0].content).toBe("signal");
}
});
});

View file

@ -40,16 +40,13 @@ 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 { 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,7 +65,7 @@ function readTermination(props: UnknownRecord): CommandTermination {
return CommandTermination.EXITED;
}
function turnsFromEvents(events: EventEnvelope[], stageId: string): TurnType[] {
export function eventsToActivity(events: EventEnvelope[], stageId: string): TurnType[] {
const stageEvents = events.filter((e) => e.node_id === stageId);
const turns: TurnType[] = [];
// Collect tool pairs: started → completed
@ -153,41 +150,6 @@ function turnsFromEvents(events: EventEnvelope[], stageId: string): TurnType[] {
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 +567,13 @@ 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 stageEventsQuery = useRunStageEvents(id, selectedStage?.id);
const turns = useMemo(
() => mapTurns(turnsQuery.data, eventsQuery.data, selectedStage?.id),
[eventsQuery.data, selectedStage?.id, turnsQuery.data],
() =>
selectedStage
? eventsToActivity(stageEventsQuery.data ?? [], selectedStage.id)
: [],
[stageEventsQuery.data, selectedStage],
);
const isRunning = selectedStage?.status === "running";
@ -661,4 +622,4 @@ export default function RunStages() {
</div>
</div>
);
}
}

View file

@ -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 workflow node (stage).
parameters:
- $ref: "#/components/parameters/RunId"
- $ref: "#/components/parameters/StageId"
- $ref: "#/components/parameters/PageLimit"
- $ref: "#/components/parameters/PageOffset"
- $ref: "#/components/parameters/StageNodeId"
- $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"
@ -2931,6 +2931,15 @@ components:
type: string
example: code@2
StageNodeId:
name: stageId
in: path
required: true
description: Workflow node id (matches RunStage.id; not visit-qualified).
schema:
type: string
example: detect-drift
CommandLogStream:
name: stream
in: path
@ -3856,20 +3865,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
@ -6340,103 +6335,6 @@ components:
description: Node identifier in the Graphviz graph source.
example: propose
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:
@ -8323,4 +8221,4 @@ components:
login:
type: string
description: User's login identifier (e.g. GitHub username).
example: octocat
example: octocat

View file

@ -15,8 +15,9 @@ 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;
@ -133,13 +134,50 @@ pub(crate) async fn get_run_stages(
paginated_response(runs::stages(), &pagination)
}
pub(crate) async fn get_stage_turns(
#[derive(serde::Deserialize)]
pub(crate) struct DemoEventListParams {
#[serde(default)]
since_seq: Option<u32>,
#[serde(default)]
limit: Option<usize>,
}
impl DemoEventListParams {
fn since_seq(&self) -> u32 {
self.since_seq.unwrap_or(1).max(1)
}
fn limit(&self) -> usize {
self.limit.unwrap_or(100).clamp(1, 1000)
}
}
pub(crate) async fn get_stage_events(
_auth: RequiredUser,
State(_state): State<Arc<AppState>>,
Path((_id, _stage_id)): Path<(String, String)>,
Query(pagination): Query<PaginationParams>,
Path((_id, stage_id)): Path<(String, String)>,
Query(params): Query<DemoEventListParams>,
) -> Response {
paginated_response(runs::turns(), &pagination)
let since_seq = params.since_seq();
let limit = params.limit();
let mut matches: Vec<EventEnvelope> = runs::stage_events()
.into_iter()
.filter(|envelope| {
envelope.seq >= since_seq
&& envelope.event.node_id.as_deref() == Some(stage_id.as_str())
})
.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(
@ -1212,18 +1250,113 @@ mod runs {
]
}
pub(super) fn turns() -> Vec<StageTurn> {
pub(super) fn stage_events() -> Vec<fabro_types::EventEnvelope> {
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 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: None,
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,
}),
),
]
}

View file

@ -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<Arc<AppState>> for RequireRunBlob {
}
}
impl FromRequestParts<Arc<AppState>> for RequireRunStageScoped {
type Rejection = Response;
async fn from_request_parts(
parts: &mut Parts,
state: &Arc<AppState>,
) -> Result<Self, Self::Rejection> {
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<Arc<AppState>> for RequireStageArtifact {
type Rejection = Response;

View file

@ -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};

View file

@ -3,9 +3,9 @@ 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, redact_jsonl_line, reject_if_archived, update_live_run_from_event,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
@ -15,6 +15,10 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
"/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))
}
@ -200,6 +204,35 @@ async fn list_run_events(
}
}
async fn list_run_stage_events(
RequireRunStageScoped(id, stage_id): RequireRunStageScoped,
State(state): State<Arc<AppState>>,
Query(params): Query<EventListParams>,
) -> 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_node_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<Arc<AppState>>,
@ -342,3 +375,202 @@ 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<Body> {
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 {
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));
}
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/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<u64> = 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/events?since_seq=203"
)))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response).await;
let seqs: Vec<u64> = 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/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/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/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/events"))
.header(header::ACCEPT, "application/json")
.body(Body::empty())
.unwrap();
let response = app.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
}

View file

@ -57,8 +57,8 @@ pub(super) fn demo_routes() -> Router<Arc<AppState>> {
.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<Arc<AppState>> {
pub(super) fn real_routes() -> Router<Arc<AppState>> {
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))

View file

@ -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/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/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/events?since_seq=4",
)
.await;
let data = body["data"].as_array().expect("data is an array");
assert_eq!(data.len(), 4);
let seqs: Vec<u64> = 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);
}

View file

@ -4,6 +4,7 @@
)]
mod api;
mod event_pagination;
mod helpers;
mod openapi_conformance;
mod pagination;

View file

@ -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",

View file

@ -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 workflow node,
/// starting at `start_seq`. The `+1` lets callers compute `has_more`.
///
/// Implementation note: scans the unbounded run-event prefix and
/// filters by `node_id` *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_node_from_with_limit(
&self,
node_id: &str,
start_seq: u32,
limit: usize,
) -> Result<Vec<EventEnvelope>> {
list_events_for_node_from_with_limit(
&self.inner.db,
&self.inner.run_id,
node_id,
start_seq,
limit,
)
.await
}
pub fn watch_events_from(
&self,
seq: u32,
@ -348,6 +372,40 @@ where
Ok(events)
}
async fn list_events_for_node_from_with_limit<R>(
db: &R,
run_id: &RunId,
node_id: &str,
start_seq: u32,
limit: usize,
) -> Result<Vec<EventEnvelope>>
where
R: DbRead + Sync,
{
// Unbounded scan first: filtering by node_id with a generic
// limit-bounded scan would silently drop matches whenever the stage's
// events are sparse late in the event log.
let mut iter = db.scan_prefix(keys::run_events_prefix(run_id)).await?;
let mut events: Vec<EventEnvelope> = Vec::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
let Some(seq) = keys::parse_event_seq(&key) else {
continue;
};
if seq < start_seq {
continue;
}
let event: RunEvent = serde_json::from_slice(&entry.value)?;
if event.node_id.as_deref() != Some(node_id) {
continue;
}
events.push(EventEnvelope { seq, event });
}
events.sort_by_key(|event| event.seq);
events.truncate(limit.saturating_add(1));
Ok(events)
}
async fn list_blobs<R>(db: &R) -> Result<Vec<RunBlobId>>
where
R: DbRead + Sync,
@ -375,9 +433,12 @@ mod tests {
use std::sync::Arc;
use std::time::Duration;
use fabro_types::RunId;
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 +455,143 @@ 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 {
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));
}
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_node_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_node_from_with_limit("alpha", 1, 100)
.await
.unwrap();
let seqs: Vec<u32> = events.iter().map(|e| e.seq).collect();
assert_eq!(seqs, vec![1, 3]);
}
#[tokio::test]
async fn list_events_for_node_skips_events_with_no_node_id() {
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_node_from_with_limit("alpha", 1, 100)
.await
.unwrap();
let seqs: Vec<u32> = events.iter().map(|e| e.seq).collect();
assert_eq!(seqs, vec![2]);
}
#[tokio::test]
async fn list_events_for_node_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_node_from_with_limit("alpha", 2, 100)
.await
.unwrap();
let seqs: Vec<u32> = events.iter().map(|e| e.seq).collect();
assert_eq!(seqs, vec![3, 5]);
}
#[tokio::test]
async fn list_events_for_node_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_node_from_with_limit("alpha", 1, 5)
.await
.unwrap();
let seqs: Vec<u32> = events.iter().map(|e| e.seq).collect();
assert_eq!(seqs, vec![201, 202, 203]);
}
#[tokio::test]
async fn list_events_for_node_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_node_from_with_limit("alpha", 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);
}
}

View file

@ -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

View file

@ -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 workflow node (stage).
* @summary List Stage Events
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {number} [pageLimit] Maximum number of items to return per page.
* @param {number} [pageOffset] Number of items to skip before returning results.
* @param {string} stageId Workflow node id (matches RunStage.id; not visit-qualified).
* @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<RequestArgs> => {
listStageEvents: async (id: string, stageId: string, sinceSeq?: number, limit?: number, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// 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 workflow node (stage).
* @summary List Stage Events
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {number} [pageLimit] Maximum number of items to return per page.
* @param {number} [pageOffset] Number of items to skip before returning results.
* @param {string} stageId Workflow node id (matches RunStage.id; not visit-qualified).
* @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<PaginatedStageTurnList>> {
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<PaginatedEventList>> {
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 workflow node (stage).
* @summary List Stage Events
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {number} [pageLimit] Maximum number of items to return per page.
* @param {number} [pageOffset] Number of items to skip before returning results.
* @param {string} stageId Workflow node id (matches RunStage.id; not visit-qualified).
* @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<PaginatedStageTurnList> {
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<PaginatedEventList> {
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 workflow node (stage).
* @summary List Stage Events
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {number} [pageLimit] Maximum number of items to return per page.
* @param {number} [pageOffset] Number of items to skip before returning results.
* @param {string} stageId Workflow node id (matches RunStage.id; not visit-qualified).
* @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));
}
/**

View file

@ -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];

View file

@ -21,3 +21,6 @@ export interface BoardColumnDefinition {
'id': BoardColumn;
'name': string;
}

View file

@ -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';

View file

@ -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;
}

View file

@ -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];

View file

@ -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<StageTurn>;
'meta': PaginationMeta;
}

View file

@ -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;
}

View file

@ -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;
}

View file

@ -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;

View file

@ -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];

View file

@ -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<ToolUse>;
}
export const ToolStageTurnKindEnum = {
TOOL: 'tool'
} as const;
export type ToolStageTurnKindEnum = typeof ToolStageTurnKindEnum[keyof typeof ToolStageTurnKindEnum];

View file

@ -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;
}