mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-09 03:20:56 +00:00
feat(events): add backward cursor pagination
This commit is contained in:
parent
e3011481a1
commit
bb1afae363
8 changed files with 574 additions and 40 deletions
|
|
@ -2838,11 +2838,17 @@ paths:
|
|||
operationId: listRunEvents
|
||||
tags: [Run Internals]
|
||||
summary: List Run Events
|
||||
description: Returns a paginated JSON list of stored run events.
|
||||
description: |
|
||||
Returns a paginated JSON list of stored run events. Ascending order
|
||||
uses `since_seq` as an inclusive cursor. Descending order uses
|
||||
`before_seq` as an exclusive cursor and starts at the newest event
|
||||
when `before_seq` is omitted.
|
||||
parameters:
|
||||
- $ref: "#/components/parameters/RunId"
|
||||
- $ref: "#/components/parameters/SinceSeq"
|
||||
- $ref: "#/components/parameters/EventLimit"
|
||||
- $ref: "#/components/parameters/BeforeSeq"
|
||||
- $ref: "#/components/parameters/EventOrder"
|
||||
responses:
|
||||
"200":
|
||||
description: Paginated list of run events
|
||||
|
|
@ -2850,6 +2856,15 @@ paths:
|
|||
application/json:
|
||||
schema:
|
||||
$ref: "#/components/schemas/PaginatedEventList"
|
||||
"400":
|
||||
description: Invalid cursor and order combination
|
||||
headers:
|
||||
x-request-id:
|
||||
$ref: "#/components/headers/XRequestId"
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
$ref: "#/components/schemas/ErrorResponse"
|
||||
"404":
|
||||
description: Run not found
|
||||
headers:
|
||||
|
|
@ -5853,6 +5868,31 @@ components:
|
|||
default: 1
|
||||
example: 42
|
||||
|
||||
BeforeSeq:
|
||||
name: before_seq
|
||||
in: query
|
||||
required: false
|
||||
description: |
|
||||
Exclusive upper event sequence cursor for descending order. Omit on
|
||||
the first descending request to start from the newest event.
|
||||
schema:
|
||||
type: integer
|
||||
minimum: 1
|
||||
example: 42
|
||||
|
||||
EventOrder:
|
||||
name: order
|
||||
in: query
|
||||
required: false
|
||||
description: |
|
||||
Event sequence order. `since_seq` is valid only with `asc`;
|
||||
`before_seq` is valid only with `desc`.
|
||||
schema:
|
||||
type: string
|
||||
enum: [asc, desc]
|
||||
default: asc
|
||||
example: desc
|
||||
|
||||
EventLimit:
|
||||
name: limit
|
||||
in: query
|
||||
|
|
|
|||
|
|
@ -42,10 +42,14 @@ pub(crate) async fn run(
|
|||
None => None,
|
||||
};
|
||||
|
||||
let events = client
|
||||
.list_run_events(&run_id, None, None)
|
||||
.await
|
||||
.context("Failed to list server-backed run events")?;
|
||||
let events = match (args.tail, since_cutoff.is_none()) {
|
||||
(Some(tail), true) => {
|
||||
let tail = if args.follow { tail.max(1) } else { tail };
|
||||
client.list_run_events_tail(&run_id, tail).await
|
||||
}
|
||||
_ => client.list_run_events(&run_id, None, None).await,
|
||||
}
|
||||
.context("Failed to list server-backed run events")?;
|
||||
let last_seq = events.last().map_or(0, |event| event.seq);
|
||||
let all_lines = events
|
||||
.iter()
|
||||
|
|
|
|||
|
|
@ -30,12 +30,24 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
|
|||
.route("/runs/{id}/attach", get(attach_run_events))
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
enum EventSequenceOrder {
|
||||
#[default]
|
||||
Asc,
|
||||
Desc,
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
pub(crate) struct EventListParams {
|
||||
#[serde(default)]
|
||||
since_seq: Option<u32>,
|
||||
since_seq: Option<u32>,
|
||||
#[serde(default)]
|
||||
limit: Option<usize>,
|
||||
before_seq: Option<u32>,
|
||||
#[serde(default)]
|
||||
order: EventSequenceOrder,
|
||||
#[serde(default)]
|
||||
limit: Option<usize>,
|
||||
}
|
||||
|
||||
impl EventListParams {
|
||||
|
|
@ -46,6 +58,26 @@ impl EventListParams {
|
|||
pub(crate) fn limit(&self) -> usize {
|
||||
self.limit.unwrap_or(100).clamp(1, 1000)
|
||||
}
|
||||
|
||||
fn before_seq(&self) -> Option<u32> {
|
||||
self.before_seq.map(|seq| seq.max(1))
|
||||
}
|
||||
|
||||
fn order(&self) -> EventSequenceOrder {
|
||||
self.order
|
||||
}
|
||||
|
||||
fn cursor_error(&self) -> Option<&'static str> {
|
||||
match self.order {
|
||||
EventSequenceOrder::Asc if self.before_seq.is_some() => {
|
||||
Some("before_seq requires order=desc.")
|
||||
}
|
||||
EventSequenceOrder::Desc if self.since_seq.is_some() => {
|
||||
Some("since_seq cannot be combined with order=desc; use before_seq instead.")
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
|
|
@ -202,29 +234,43 @@ async fn list_run_events(
|
|||
State(state): State<Arc<AppState>>,
|
||||
Query(params): Query<EventListParams>,
|
||||
) -> Response {
|
||||
if let Some(detail) = params.cursor_error() {
|
||||
return ApiError::bad_request(detail).into_response();
|
||||
}
|
||||
|
||||
let since_seq = params.since_seq();
|
||||
let limit = params.limit();
|
||||
match state.stores.runs.open_run_reader(&id).await {
|
||||
Ok(run_store) => match run_store
|
||||
.list_events_from_with_limit(since_seq, limit)
|
||||
.await
|
||||
{
|
||||
Ok(mut events) => {
|
||||
let has_more = events.len() > limit;
|
||||
events.truncate(limit);
|
||||
Json(PaginatedEventList {
|
||||
data: events,
|
||||
meta: PaginationMeta {
|
||||
has_more,
|
||||
total: None,
|
||||
},
|
||||
})
|
||||
.into_response()
|
||||
Ok(run_store) => {
|
||||
let events = match params.order() {
|
||||
EventSequenceOrder::Asc => {
|
||||
run_store
|
||||
.list_events_from_with_limit(since_seq, limit)
|
||||
.await
|
||||
}
|
||||
EventSequenceOrder::Desc => {
|
||||
run_store
|
||||
.list_events_before_with_limit(params.before_seq(), limit)
|
||||
.await
|
||||
}
|
||||
};
|
||||
match events {
|
||||
Ok(mut events) => {
|
||||
let has_more = events.len() > limit;
|
||||
events.truncate(limit);
|
||||
Json(PaginatedEventList {
|
||||
data: events,
|
||||
meta: PaginationMeta {
|
||||
has_more,
|
||||
total: None,
|
||||
},
|
||||
})
|
||||
.into_response()
|
||||
}
|
||||
Err(err) => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
|
||||
.into_response(),
|
||||
}
|
||||
Err(err) => {
|
||||
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
|
||||
}
|
||||
},
|
||||
}
|
||||
Err(_) => ApiError::not_found("Run not found.").into_response(),
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9828,6 +9828,97 @@ async fn list_run_events_returns_paginated_json() {
|
|||
assert!(body["meta"]["has_more"].is_boolean());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_run_events_descends_from_latest_with_exclusive_cursor() {
|
||||
let state = test_app_state();
|
||||
let app = crate::test_support::build_test_router(Arc::clone(&state));
|
||||
let run_id = RunId::new();
|
||||
create_durable_run_with_events(&state, run_id, &[
|
||||
workflow_event::Event::RunRunnable {
|
||||
source: fabro_types::RunRunnableSource::StartRequested,
|
||||
actor: None,
|
||||
},
|
||||
workflow_event::Event::RunStarting,
|
||||
workflow_event::Event::RunRunning,
|
||||
])
|
||||
.await;
|
||||
|
||||
let response = app
|
||||
.clone()
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("GET")
|
||||
.uri(api(&format!("/runs/{run_id}/events?order=desc&limit=2")))
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let body = response_json!(response, StatusCode::OK).await;
|
||||
let seqs = body["data"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|event| event["seq"].as_u64().unwrap())
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(seqs, vec![4, 3]);
|
||||
assert_eq!(body["meta"]["has_more"], true);
|
||||
|
||||
let response = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("GET")
|
||||
.uri(api(&format!(
|
||||
"/runs/{run_id}/events?order=desc&before_seq=3&limit=2"
|
||||
)))
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let body = response_json!(response, StatusCode::OK).await;
|
||||
let seqs = body["data"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|event| event["seq"].as_u64().unwrap())
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(seqs, vec![2, 1]);
|
||||
assert_eq!(body["meta"]["has_more"], false);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_run_events_rejects_cursor_for_opposite_order() {
|
||||
let app = crate::test_support::build_test_router(test_app_state());
|
||||
let run_id = RunId::new();
|
||||
let cases = [
|
||||
(
|
||||
format!("/runs/{run_id}/events?order=desc&since_seq=2"),
|
||||
"since_seq cannot be combined with order=desc; use before_seq instead.",
|
||||
),
|
||||
(
|
||||
format!("/runs/{run_id}/events?before_seq=2"),
|
||||
"before_seq requires order=desc.",
|
||||
),
|
||||
];
|
||||
|
||||
for (path, expected_detail) in cases {
|
||||
let response = app
|
||||
.clone()
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.method("GET")
|
||||
.uri(api(&path))
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let body = response_json!(response, StatusCode::BAD_REQUEST).await;
|
||||
assert_eq!(body["errors"][0]["detail"], expected_detail);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn append_run_event_rejects_run_id_mismatch() {
|
||||
let state = test_app_state();
|
||||
|
|
|
|||
|
|
@ -171,6 +171,15 @@ impl RunProjectionCache {
|
|||
.map(|entry| state.with_children_count(entry))
|
||||
}
|
||||
|
||||
pub(crate) async fn last_seq(&self, run_id: &RunId) -> Option<u32> {
|
||||
self.state
|
||||
.lock()
|
||||
.await
|
||||
.entries
|
||||
.get(run_id)
|
||||
.map(|entry| entry.last_seq)
|
||||
}
|
||||
|
||||
pub(crate) async fn get_summary(&self, run_id: &RunId, now: DateTime<Utc>) -> Option<Run> {
|
||||
let mut entry = {
|
||||
let state = self.state.lock().await;
|
||||
|
|
|
|||
|
|
@ -379,6 +379,58 @@ impl RunDatabase {
|
|||
list_events_from_with_limit(&self.inner.db, &self.inner.run_id, start_seq, limit).await
|
||||
}
|
||||
|
||||
/// Returns up to `limit + 1` events immediately before `before_seq` in
|
||||
/// descending sequence order. Omitting `before_seq` starts at the newest
|
||||
/// event. The extra item lets callers compute `has_more`.
|
||||
pub async fn list_events_before_with_limit(
|
||||
&self,
|
||||
before_seq: Option<u32>,
|
||||
limit: usize,
|
||||
) -> Result<Vec<EventEnvelope>> {
|
||||
let latest_seq = match self
|
||||
.inner
|
||||
.shared_projection_cache
|
||||
.last_seq(&self.inner.run_id)
|
||||
.await
|
||||
{
|
||||
Some(seq) => seq,
|
||||
None => recover_next_seq(
|
||||
&self.inner.db,
|
||||
keys::run_events_prefix(&self.inner.run_id),
|
||||
keys::parse_event_seq,
|
||||
)
|
||||
.await?
|
||||
.saturating_sub(1),
|
||||
};
|
||||
if latest_seq == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let latest_exclusive = u64::from(latest_seq) + 1;
|
||||
let end_seq = before_seq
|
||||
.map_or(latest_exclusive, u64::from)
|
||||
.min(latest_exclusive)
|
||||
.max(1);
|
||||
if end_seq == 1 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let window_size = u64::try_from(limit.saturating_add(1)).unwrap_or(u64::MAX);
|
||||
let start_seq =
|
||||
u32::try_from(end_seq.saturating_sub(window_size).max(1)).unwrap_or(u32::MAX);
|
||||
let end_seq = u32::try_from(end_seq).ok();
|
||||
let mut events = list_events_in_range_with_limit(
|
||||
&self.inner.db,
|
||||
&self.inner.run_id,
|
||||
start_seq,
|
||||
end_seq,
|
||||
limit,
|
||||
)
|
||||
.await?;
|
||||
events.reverse();
|
||||
Ok(events)
|
||||
}
|
||||
|
||||
pub async fn get_event(&self, seq: u32) -> Result<Option<EventEnvelope>> {
|
||||
get_event(&self.inner.db, &self.inner.run_id, seq).await
|
||||
}
|
||||
|
|
@ -568,13 +620,35 @@ async fn list_events_from_with_limit<R>(
|
|||
where
|
||||
R: DbRead + Sync,
|
||||
{
|
||||
list_events_in_range_with_limit(db, run_id, start_seq, None, limit).await
|
||||
}
|
||||
|
||||
async fn list_events_in_range_with_limit<R>(
|
||||
db: &R,
|
||||
run_id: &RunId,
|
||||
start_seq: u32,
|
||||
end_seq: Option<u32>,
|
||||
limit: usize,
|
||||
) -> Result<Vec<EventEnvelope>>
|
||||
where
|
||||
R: DbRead + Sync,
|
||||
{
|
||||
if end_seq.is_some_and(|end_seq| end_seq <= start_seq) {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let event_prefix = keys::run_events_prefix(run_id);
|
||||
let max_events = limit.saturating_add(1);
|
||||
// Seek to the page cursor and decode only the requested page plus the
|
||||
// sentinel used to compute `has_more`.
|
||||
let mut iter = db
|
||||
.scan(keys::run_event_seq_prefix(run_id, start_seq)..)
|
||||
.await?;
|
||||
let start_key = keys::run_event_seq_prefix(run_id, start_seq);
|
||||
let mut iter = match end_seq {
|
||||
Some(end_seq) => {
|
||||
let end_key = keys::run_event_seq_prefix(run_id, end_seq);
|
||||
db.scan(start_key..end_key).await?
|
||||
}
|
||||
None => db.scan(start_key..).await?,
|
||||
};
|
||||
let mut events = Vec::new();
|
||||
while events.len() < max_events {
|
||||
let Some(entry) = iter.next().await? else {
|
||||
|
|
@ -916,6 +990,65 @@ mod tests {
|
|||
assert_eq!(seqs, vec![3]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_events_before_with_limit_returns_newest_events_and_sentinel() {
|
||||
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_before_with_limit(None, 2).await.unwrap();
|
||||
|
||||
let seqs: Vec<u32> = events.iter().map(|event| event.seq).collect();
|
||||
assert_eq!(seqs, vec![6, 5, 4]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_events_before_with_limit_does_not_read_older_history() {
|
||||
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();
|
||||
}
|
||||
run.inner
|
||||
.db
|
||||
.put(keys::run_event_key(&run_id, 2, 0), b"invalid json")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let events = run.list_events_before_with_limit(None, 2).await.unwrap();
|
||||
|
||||
let seqs: Vec<u32> = events.iter().map(|event| event.seq).collect();
|
||||
assert_eq!(seqs, vec![6, 5, 4]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_events_before_with_limit_uses_exclusive_cursor() {
|
||||
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_before_with_limit(Some(5), 2).await.unwrap();
|
||||
|
||||
let seqs: Vec<u32> = events.iter().map(|event| event.seq).collect();
|
||||
assert_eq!(seqs, vec![4, 3, 2]);
|
||||
assert!(
|
||||
run.list_events_before_with_limit(Some(1), 2)
|
||||
.await
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_events_for_stage_returns_only_matching_events_in_seq_order() {
|
||||
let run = fresh_run().await;
|
||||
|
|
|
|||
|
|
@ -1564,6 +1564,74 @@ impl Client {
|
|||
Ok(all_events)
|
||||
}
|
||||
|
||||
/// Returns the newest `max_events` in ascending sequence order.
|
||||
pub async fn list_run_events_tail(
|
||||
&self,
|
||||
run_id: &RunId,
|
||||
max_events: usize,
|
||||
) -> Result<Vec<EventEnvelope>> {
|
||||
if max_events == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
// Fetch two events when the caller asks for one so an older server
|
||||
// that silently ignores the new order parameter can be detected.
|
||||
let fetch_target = max_events.max(2);
|
||||
let mut before_seq = None;
|
||||
let mut descending_events: Vec<EventEnvelope> = Vec::new();
|
||||
while descending_events.len() < fetch_target {
|
||||
let remaining = fetch_target - descending_events.len();
|
||||
let response = self
|
||||
.send_api(|client| async move {
|
||||
let mut request = client
|
||||
.list_run_events()
|
||||
.id(run_id.to_string())
|
||||
.order(types::ListRunEventsOrder::Desc)
|
||||
.limit(remaining.min(1000) as u64);
|
||||
if let Some(seq) = before_seq.and_then(non_zero_u64_from_u32) {
|
||||
request = request.before_seq(seq);
|
||||
}
|
||||
request.send().await
|
||||
})
|
||||
.await?;
|
||||
let parsed = response.into_inner();
|
||||
let page_events = parsed
|
||||
.data
|
||||
.into_iter()
|
||||
.map(convert_type::<_, EventEnvelope>)
|
||||
.collect::<Result<Vec<EventEnvelope>>>()?;
|
||||
|
||||
let page_is_descending = page_events
|
||||
.windows(2)
|
||||
.all(|events| events[0].seq > events[1].seq);
|
||||
let continues_descending = descending_events
|
||||
.last()
|
||||
.zip(page_events.first())
|
||||
.is_none_or(|(previous, next)| previous.seq > next.seq);
|
||||
if !page_is_descending || !continues_descending {
|
||||
let mut events = self.list_run_events(run_id, None, None).await?;
|
||||
let tail_start = events.len().saturating_sub(max_events);
|
||||
return Ok(events.split_off(tail_start));
|
||||
}
|
||||
|
||||
let next_before_seq = page_events.last().map(|event| event.seq);
|
||||
let had_events = !page_events.is_empty();
|
||||
descending_events.extend(page_events);
|
||||
if descending_events.len() >= fetch_target
|
||||
|| !parsed.meta.has_more
|
||||
|| !had_events
|
||||
|| next_before_seq.is_none()
|
||||
{
|
||||
break;
|
||||
}
|
||||
before_seq = next_before_seq;
|
||||
}
|
||||
|
||||
descending_events.reverse();
|
||||
let tail_start = descending_events.len().saturating_sub(max_events);
|
||||
Ok(descending_events.split_off(tail_start))
|
||||
}
|
||||
|
||||
pub async fn list_run_events_until(
|
||||
&self,
|
||||
run_id: &RunId,
|
||||
|
|
@ -2148,6 +2216,17 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
fn run_event_json(run_id: &RunId, seq: u32) -> serde_json::Value {
|
||||
json!({
|
||||
"seq": seq,
|
||||
"event": "run.running",
|
||||
"id": format!("evt-{seq}"),
|
||||
"run_id": run_id,
|
||||
"ts": "2026-07-24T12:00:00Z",
|
||||
"properties": {},
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn refresh_access_token_allows_plain_http_targets() {
|
||||
|
|
@ -2280,6 +2359,116 @@ mod tests {
|
|||
assert!(models.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_run_events_tail_pages_backward_and_returns_ascending() {
|
||||
let server = MockServer::start_async().await;
|
||||
let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
|
||||
let newest_page = server
|
||||
.mock_async(|when, then| {
|
||||
when.method(GET)
|
||||
.path(format!("/api/v1/runs/{run_id}/events"))
|
||||
.query_param("order", "desc")
|
||||
.query_param("limit", "5");
|
||||
then.status(200)
|
||||
.header("Content-Type", "application/json")
|
||||
.json_body(json!({
|
||||
"data": [
|
||||
run_event_json(&run_id, 6),
|
||||
run_event_json(&run_id, 5),
|
||||
run_event_json(&run_id, 4),
|
||||
],
|
||||
"meta": { "has_more": true },
|
||||
}));
|
||||
})
|
||||
.await;
|
||||
let older_page = server
|
||||
.mock_async(|when, then| {
|
||||
when.method(GET)
|
||||
.path(format!("/api/v1/runs/{run_id}/events"))
|
||||
.query_param("order", "desc")
|
||||
.query_param("before_seq", "4")
|
||||
.query_param("limit", "2");
|
||||
then.status(200)
|
||||
.header("Content-Type", "application/json")
|
||||
.json_body(json!({
|
||||
"data": [
|
||||
run_event_json(&run_id, 3),
|
||||
run_event_json(&run_id, 2),
|
||||
],
|
||||
"meta": { "has_more": true },
|
||||
}));
|
||||
})
|
||||
.await;
|
||||
|
||||
let client = Client::new_no_proxy(&server.url("")).unwrap();
|
||||
let events = client.list_run_events_tail(&run_id, 5).await.unwrap();
|
||||
|
||||
newest_page.assert_async().await;
|
||||
older_page.assert_async().await;
|
||||
let seqs = events
|
||||
.into_iter()
|
||||
.map(|event| event.seq)
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(seqs, vec![2, 3, 4, 5, 6]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_run_events_tail_falls_back_when_server_ignores_descending_order() {
|
||||
let server = MockServer::start_async().await;
|
||||
let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
|
||||
let unsupported_descending_page = server
|
||||
.mock_async(|when, then| {
|
||||
when.method(GET)
|
||||
.path(format!("/api/v1/runs/{run_id}/events"))
|
||||
.query_param("order", "desc")
|
||||
.query_param("limit", "3");
|
||||
then.status(200)
|
||||
.header("Content-Type", "application/json")
|
||||
.json_body(json!({
|
||||
"data": [
|
||||
run_event_json(&run_id, 1),
|
||||
run_event_json(&run_id, 2),
|
||||
run_event_json(&run_id, 3),
|
||||
],
|
||||
"meta": { "has_more": true },
|
||||
}));
|
||||
})
|
||||
.await;
|
||||
let full_history = server
|
||||
.mock_async(|when, then| {
|
||||
when.method(GET)
|
||||
.path(format!("/api/v1/runs/{run_id}/events"))
|
||||
.query_param_missing("order")
|
||||
.query_param_missing("before_seq")
|
||||
.query_param_missing("since_seq")
|
||||
.query_param_missing("limit");
|
||||
then.status(200)
|
||||
.header("Content-Type", "application/json")
|
||||
.json_body(json!({
|
||||
"data": [
|
||||
run_event_json(&run_id, 1),
|
||||
run_event_json(&run_id, 2),
|
||||
run_event_json(&run_id, 3),
|
||||
run_event_json(&run_id, 4),
|
||||
run_event_json(&run_id, 5),
|
||||
],
|
||||
"meta": { "has_more": false },
|
||||
}));
|
||||
})
|
||||
.await;
|
||||
|
||||
let client = Client::new_no_proxy(&server.url("")).unwrap();
|
||||
let events = client.list_run_events_tail(&run_id, 3).await.unwrap();
|
||||
|
||||
unsupported_descending_page.assert_async().await;
|
||||
full_history.assert_async().await;
|
||||
let seqs = events
|
||||
.into_iter()
|
||||
.map(|event| event.seq)
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(seqs, vec![3, 4, 5]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_provider_credentials_posts_api_key() {
|
||||
let server = MockServer::start_async().await;
|
||||
|
|
|
|||
|
|
@ -472,15 +472,17 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
|
|||
};
|
||||
},
|
||||
/**
|
||||
* Returns a paginated JSON list of stored run events.
|
||||
* Returns a paginated JSON list of stored run events. Ascending order uses `since_seq` as an inclusive cursor. Descending order uses `before_seq` as an exclusive cursor and starts at the newest event when `before_seq` is omitted.
|
||||
* @summary List Run Events
|
||||
* @param {string} id Unique run identifier (ULID).
|
||||
* @param {number} [sinceSeq] First event sequence number to include.
|
||||
* @param {number} [limit] Maximum number of events to return.
|
||||
* @param {number} [beforeSeq] Exclusive upper event sequence cursor for descending order. Omit on the first descending request to start from the newest event.
|
||||
* @param {ListRunEventsOrderEnum} [order] Event sequence order. `since_seq` is valid only with `asc`; `before_seq` is valid only with `desc`.
|
||||
* @param {*} [options] Override http request option.
|
||||
* @throws {RequiredError}
|
||||
*/
|
||||
listRunEvents: async (id: string, sinceSeq?: number, limit?: number, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
|
||||
listRunEvents: async (id: string, sinceSeq?: number, limit?: number, beforeSeq?: number, order?: ListRunEventsOrderEnum, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
|
||||
// verify required parameter 'id' is not null or undefined
|
||||
assertParamExists('listRunEvents', 'id', id)
|
||||
const localVarPath = `/api/v1/runs/{id}/events`
|
||||
|
|
@ -510,6 +512,14 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
|
|||
localVarQueryParameter['limit'] = limit;
|
||||
}
|
||||
|
||||
if (beforeSeq !== undefined) {
|
||||
localVarQueryParameter['before_seq'] = beforeSeq;
|
||||
}
|
||||
|
||||
if (order !== undefined) {
|
||||
localVarQueryParameter['order'] = order;
|
||||
}
|
||||
|
||||
localVarHeaderParameter['Accept'] = 'application/json';
|
||||
|
||||
setSearchParams(localVarUrlObj, localVarQueryParameter);
|
||||
|
|
@ -1037,16 +1047,18 @@ export const RunInternalsApiFp = function(configuration?: Configuration) {
|
|||
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
|
||||
},
|
||||
/**
|
||||
* Returns a paginated JSON list of stored run events.
|
||||
* Returns a paginated JSON list of stored run events. Ascending order uses `since_seq` as an inclusive cursor. Descending order uses `before_seq` as an exclusive cursor and starts at the newest event when `before_seq` is omitted.
|
||||
* @summary List Run Events
|
||||
* @param {string} id Unique run identifier (ULID).
|
||||
* @param {number} [sinceSeq] First event sequence number to include.
|
||||
* @param {number} [limit] Maximum number of events to return.
|
||||
* @param {number} [beforeSeq] Exclusive upper event sequence cursor for descending order. Omit on the first descending request to start from the newest event.
|
||||
* @param {ListRunEventsOrderEnum} [order] Event sequence order. `since_seq` is valid only with `asc`; `before_seq` is valid only with `desc`.
|
||||
* @param {*} [options] Override http request option.
|
||||
* @throws {RequiredError}
|
||||
*/
|
||||
async listRunEvents(id: string, sinceSeq?: number, limit?: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<PaginatedEventList>> {
|
||||
const localVarAxiosArgs = await localVarAxiosParamCreator.listRunEvents(id, sinceSeq, limit, options);
|
||||
async listRunEvents(id: string, sinceSeq?: number, limit?: number, beforeSeq?: number, order?: ListRunEventsOrderEnum, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<PaginatedEventList>> {
|
||||
const localVarAxiosArgs = await localVarAxiosParamCreator.listRunEvents(id, sinceSeq, limit, beforeSeq, order, options);
|
||||
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
|
||||
const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.listRunEvents']?.[localVarOperationServerIndex]?.url;
|
||||
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
|
||||
|
|
@ -1278,16 +1290,18 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b
|
|||
return localVarFp.listRunArtifacts(id, options).then((request) => request(axios, basePath));
|
||||
},
|
||||
/**
|
||||
* Returns a paginated JSON list of stored run events.
|
||||
* Returns a paginated JSON list of stored run events. Ascending order uses `since_seq` as an inclusive cursor. Descending order uses `before_seq` as an exclusive cursor and starts at the newest event when `before_seq` is omitted.
|
||||
* @summary List Run Events
|
||||
* @param {string} id Unique run identifier (ULID).
|
||||
* @param {number} [sinceSeq] First event sequence number to include.
|
||||
* @param {number} [limit] Maximum number of events to return.
|
||||
* @param {number} [beforeSeq] Exclusive upper event sequence cursor for descending order. Omit on the first descending request to start from the newest event.
|
||||
* @param {ListRunEventsOrderEnum} [order] Event sequence order. `since_seq` is valid only with `asc`; `before_seq` is valid only with `desc`.
|
||||
* @param {*} [options] Override http request option.
|
||||
* @throws {RequiredError}
|
||||
*/
|
||||
listRunEvents(id: string, sinceSeq?: number, limit?: number, options?: RawAxiosRequestConfig): AxiosPromise<PaginatedEventList> {
|
||||
return localVarFp.listRunEvents(id, sinceSeq, limit, options).then((request) => request(axios, basePath));
|
||||
listRunEvents(id: string, sinceSeq?: number, limit?: number, beforeSeq?: number, order?: ListRunEventsOrderEnum, options?: RawAxiosRequestConfig): AxiosPromise<PaginatedEventList> {
|
||||
return localVarFp.listRunEvents(id, sinceSeq, limit, beforeSeq, order, options).then((request) => request(axios, basePath));
|
||||
},
|
||||
/**
|
||||
* Returns the ordered list of stages in a run\'s workflow graph with their current status and timing. Stages are bounded by the workflow graph size, typically fewer than 20.
|
||||
|
|
@ -1499,16 +1513,18 @@ export class RunInternalsApi extends BaseAPI {
|
|||
}
|
||||
|
||||
/**
|
||||
* Returns a paginated JSON list of stored run events.
|
||||
* Returns a paginated JSON list of stored run events. Ascending order uses `since_seq` as an inclusive cursor. Descending order uses `before_seq` as an exclusive cursor and starts at the newest event when `before_seq` is omitted.
|
||||
* @summary List Run Events
|
||||
* @param {string} id Unique run identifier (ULID).
|
||||
* @param {number} [sinceSeq] First event sequence number to include.
|
||||
* @param {number} [limit] Maximum number of events to return.
|
||||
* @param {number} [beforeSeq] Exclusive upper event sequence cursor for descending order. Omit on the first descending request to start from the newest event.
|
||||
* @param {ListRunEventsOrderEnum} [order] Event sequence order. `since_seq` is valid only with `asc`; `before_seq` is valid only with `desc`.
|
||||
* @param {*} [options] Override http request option.
|
||||
* @throws {RequiredError}
|
||||
*/
|
||||
public listRunEvents(id: string, sinceSeq?: number, limit?: number, options?: RawAxiosRequestConfig) {
|
||||
return RunInternalsApiFp(this.configuration).listRunEvents(id, sinceSeq, limit, options).then((request) => request(this.axios, this.basePath));
|
||||
public listRunEvents(id: string, sinceSeq?: number, limit?: number, beforeSeq?: number, order?: ListRunEventsOrderEnum, options?: RawAxiosRequestConfig) {
|
||||
return RunInternalsApiFp(this.configuration).listRunEvents(id, sinceSeq, limit, beforeSeq, order, options).then((request) => request(this.axios, this.basePath));
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -1611,3 +1627,9 @@ export class RunInternalsApi extends BaseAPI {
|
|||
return RunInternalsApiFp(this.configuration).writeRunBlob(id, body, options).then((request) => request(this.axios, this.basePath));
|
||||
}
|
||||
}
|
||||
|
||||
export const ListRunEventsOrderEnum = {
|
||||
ASC: 'asc',
|
||||
DESC: 'desc'
|
||||
} as const;
|
||||
export type ListRunEventsOrderEnum = typeof ListRunEventsOrderEnum[keyof typeof ListRunEventsOrderEnum];
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue