diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 57810a34ff..7c24125d11 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -1576,9 +1576,12 @@ export interface components { */ RunOutcome: "success" | "warn" | "partial" | "failed" | "cancelled"; RunRecord: { + /** Format: date-time */ + cancel_requested_at?: string | null; /** Format: date-time */ ended_at?: string | null; job_key?: string | null; + kind: components["schemas"]["RunKind"]; /** Format: int64 */ last_event_storage_id: number; /** Format: date-time */ @@ -1588,8 +1591,22 @@ export interface components { node_counts_json: { [key: string]: number; }; + origin: components["schemas"]["RunOrigin"]; + /** Format: uuid */ + parent_run_id?: string | null; + /** Format: int64 */ + priority: number; /** Format: date-time */ queued_at: string; + /** @description The stored actor: `principal:` for a person, otherwise a system actor such as `system:scheduler`. */ + requested_by: string; + /** @description The display name of the requesting principal; never set for system actors. */ + requester_name?: string | null; + /** Format: uuid */ + rerun_of_run_id?: string | null; + rewrite_history: boolean; + /** Format: uuid */ + root_run_id: string; /** Format: uuid */ run_id: string; selection_json: string[]; @@ -1598,7 +1615,13 @@ export interface components { /** Format: date-time */ started_at?: string | null; status: components["schemas"]["RunStatus"]; + status_reason?: string | null; target: string; + /** Format: uuid */ + tick_id?: string | null; + /** Format: uuid */ + trigger_id?: string | null; + trigger_name?: string | null; }; /** @description The authorized selection and pinned execution inputs are immutable once requested. */ RunRequest: { diff --git a/crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs b/crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs index e3f248bb67..bddbdfa2df 100644 --- a/crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs +++ b/crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs @@ -13,8 +13,13 @@ pub(crate) async fn list_runs_in( limit: u32, ) -> Result<(ControlTransaction, Vec), ControlError> { query_records(transaction, - "SELECT run_id, location_id, target, snapshot_id, job_key, status, queued_at, launched_at, started_at, ended_at, selection_json, node_counts_json, last_event_storage_id \ - FROM sqb_runs WHERE location_id=$1 AND target=$2 ORDER BY queued_at DESC, run_id DESC LIMIT $3", + "SELECT r.run_id, r.location_id, r.target, r.snapshot_id, r.job_key, r.kind, r.origin, r.parent_run_id, r.root_run_id, \ + r.trigger_id, t.name AS trigger_name, r.tick_id, r.rerun_of_run_id, r.requested_by, p.name AS requester_name, r.priority, \ + r.rewrite_history, r.status, r.status_reason, r.queued_at, r.launched_at, r.started_at, r.ended_at, r.cancel_requested_at, \ + r.selection_json, r.node_counts_json, r.last_event_storage_id \ + FROM sqb_runs r LEFT JOIN sqb_triggers t ON t.trigger_id=r.trigger_id \ + LEFT JOIN sqb_principals p ON 'principal:' || CAST(p.principal_id AS TEXT)=r.requested_by \ + WHERE r.location_id=$1 AND r.target=$2 ORDER BY r.queued_at DESC, r.run_id DESC LIMIT $3", vec![Parameter::Id(location_id), Parameter::Text(target.into()), Parameter::Integer(i64::from(limit.min(1000)))]).await } diff --git a/crates/sqlbuild-orchestration/src/runs/models.rs b/crates/sqlbuild-orchestration/src/runs/models.rs index 7b6f9c7c2d..545660e12d 100644 --- a/crates/sqlbuild-orchestration/src/runs/models.rs +++ b/crates/sqlbuild-orchestration/src/runs/models.rs @@ -153,11 +153,27 @@ pub struct RunRecord { pub target: String, pub snapshot_id: Uuid, pub job_key: Option, + pub kind: RunKind, + pub origin: RunOrigin, + pub parent_run_id: Option, + pub root_run_id: Uuid, + pub trigger_id: Option, + pub trigger_name: Option, + pub tick_id: Option, + pub rerun_of_run_id: Option, + /// The stored actor: `principal:` for a person, otherwise a system actor such as `system:scheduler`. + pub requested_by: String, + /// The display name of the requesting principal; never set for system actors. + pub requester_name: Option, + pub priority: i64, + pub rewrite_history: bool, pub status: RunStatus, + pub status_reason: Option, pub queued_at: DateTime, pub launched_at: Option>, pub started_at: Option>, pub ended_at: Option>, + pub cancel_requested_at: Option>, pub selection_json: BTreeSet, pub node_counts_json: BTreeMap, pub last_event_storage_id: i64, diff --git a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs index f776628da0..6984b0a561 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs @@ -178,6 +178,9 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ let mut expected_work = 1; let mut expected_runs = 1; let mut last_request = Value::Null; + let mut listed_person_runs = 0; + let mut launched_person_runs = 0; + let mut listed_seed_runs = 0; for role_case in roles { for role in role_case.role.iter().copied() { assign_builtin_role( @@ -234,6 +237,45 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ page.is_array(), role_case.expected_metadata == StatusCode::OK.as_u16() ); + let rows = page.as_array().into_iter().flatten(); + let mut listed_here = 0; + for launched in rows + .clone() + .filter(|row| row["requested_by"] == json!(format!("principal:{person}"))) + { + listed_person_runs += 1; + listed_here += 1; + for (key, expected) in [ + ("kind", json!("job")), + ("origin", json!("manual")), + ("job_key", json!("orders")), + ("requester_name", json!("orders-person")), + ("root_run_id", launched["run_id"].clone()), + ("parent_run_id", Value::Null), + ("trigger_name", Value::Null), + ("rerun_of_run_id", Value::Null), + ("priority", json!(0)), + ("rewrite_history", json!(false)), + ("cancel_requested_at", Value::Null), + ] { + assert_eq!(launched[key], expected, "person-requested run {key}"); + } + } + assert_eq!( + listed_here, + launched_person_runs * usize::from(page.is_array()), + "every earlier person launch is listed" + ); + for seeded in rows.filter(|row| row["run_id"] == json!(seed_run.run_id)) { + listed_seed_runs += 1; + for (key, expected) in [ + ("kind", json!("adhoc")), + ("requested_by", json!("test-operator")), + ("requester_name", Value::Null), + ] { + assert_eq!(seeded[key], expected, "system-requested run {key}"); + } + } assert!(!page.to_string().contains("select 1 as id")); let known_reads = [ format!("{metadata}/{}/nodes", seed_run.run_id), @@ -343,6 +385,7 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ .as_u16(), role_case.expected_job ); + launched_person_runs += usize::from(role_case.expected_job == 200); for (endpoint, request_body) in [(&path[..], &body), ("/api/control/runs", &job)] { @@ -464,6 +507,14 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ json!({"work":expected_work,"runs":expected_runs,"claims":0}) ); } + assert!( + listed_person_runs > 0, + "a person-launched run was listed and checked" + ); + assert!( + listed_seed_runs > 0, + "the system-requested run was listed and checked" + ); assign_builtin_role( &replica, (&context, admin_session.identity.principal_id), diff --git a/crates/sqlbuild-orchestration/tests/sensors/test_custom.rs b/crates/sqlbuild-orchestration/tests/sensors/test_custom.rs index ae19f87689..e0fde5327c 100644 --- a/crates/sqlbuild-orchestration/tests/sensors/test_custom.rs +++ b/crates/sqlbuild-orchestration/tests/sensors/test_custom.rs @@ -18,6 +18,7 @@ use crate::platform::models::{ LocationWorkResult, }; use crate::runs::main::list_runs::list_runs; +use crate::runs::models::RunOrigin; use crate::scheduler::tests::contracts::helpers::acquire_lease; use crate::sensors::main::begin_sensor_evaluation::begin_sensor_evaluation; use crate::sensors::main::complete_custom_sensor::complete_custom_sensor; @@ -260,15 +261,35 @@ fn given_custom_sensor_work_when_results_retry_then_cursor_runs_and_outcome_comm .expect("retry commit"), 0 ); + let sensor_runs = list_runs(store, (location, "development"), 100) + .await + .expect("dedup runs"); assert_eq!( - list_runs(store, (location, "development"), 100) - .await - .expect("dedup runs") - .len(), + sensor_runs.len(), test_case.expected_runs, "{}", test_case.description ); + for sensor_run in &sensor_runs { + assert_eq!( + ( + sensor_run.origin, + sensor_run.trigger_id, + sensor_run.trigger_name.as_deref(), + sensor_run.requested_by.as_str(), + sensor_run.requester_name.as_deref(), + ), + ( + RunOrigin::Sensor, + Some(trigger.trigger_id), + Some(trigger.name.as_str()), + "system:sensor", + None, + ), + "{}", + test_case.description + ); + } let facts = evaluations(store).await; let SensorEvaluationOutcome::Completed { cursor, run_ids } = &facts.last().expect("completed").outcome diff --git a/crates/sqlbuild-orchestration/tests/sensors/test_runs.rs b/crates/sqlbuild-orchestration/tests/sensors/test_runs.rs index a7d19db4f8..1b88eb8d3b 100644 --- a/crates/sqlbuild-orchestration/tests/sensors/test_runs.rs +++ b/crates/sqlbuild-orchestration/tests/sensors/test_runs.rs @@ -12,7 +12,7 @@ use crate::declarations::main::list_triggers::list_triggers; use crate::declarations::models::TriggerChange; use crate::runs::main::list_runs::list_runs; use crate::runs::main::request_rerun::request_rerun; -use crate::runs::models::{NodeOutcome, RerunRequest, RunOutcome}; +use crate::runs::models::{NodeOutcome, RerunRequest, RunKind, RunOrigin, RunOutcome}; use crate::scheduler::tests::contracts::helpers::{acquire_lease, hold_lease}; use crate::sensors::main::{ begin_sensor_evaluation::begin_sensor_evaluation, @@ -383,6 +383,40 @@ fn given_terminal_jobs_and_groups_when_selectors_retry_then_outcomes_scope_and_r let sources = list_runs(store, (location, "development"), 500) .await .expect("all source executions"); + let rerun = sources + .iter() + .find(|record| record.run_id == retry.run_id) + .expect("rerun listed"); + assert_eq!( + ( + rerun.kind, + rerun.origin, + rerun.rerun_of_run_id, + rerun.priority, + rerun.root_run_id, + rerun.parent_run_id, + rerun.requested_by.as_str(), + rerun.requester_name.as_deref(), + ), + ( + RunKind::Group, + RunOrigin::Rerun, + Some(retry.source_run_id), + 10, + retry.run_id, + None, + "test-admin", + None, + ) + ); + let rerun_children: Vec<_> = sources + .iter() + .filter(|record| record.parent_run_id == Some(retry.run_id)) + .collect(); + assert!(!rerun_children.is_empty(), "rerun children listed"); + for child in rerun_children { + assert_eq!(child.root_run_id, retry.run_id, "child root"); + } // Same key in another target/location and a different job in this scope must not match. let mut other_input = input.clone(); other_input.location_id = other; diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index 0944f8eb5f..fd900528bf 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -97,7 +97,9 @@ export interface ApiClient extends PlatformApi { /** Writes a `cancel_requested` fact; the scheduler then signals the worker. */ cancelRun(scope: Scope, runId: string): Promise; /** `nodes` is required for `selected` and ignored otherwise. */ - rerunRun(scope: Scope, runId: string, mode: RerunMode, nodes?: string[]): Promise; + rerunRun(scope: Scope, runId: string, mode: Exclude): Promise; + /** Reruns exactly `nodes` of the run. */ + rerunSelectedNodes(scope: Scope, runId: string, nodes: string[]): Promise; listSnapshotNodes(scope: Scope): Promise; /** The server's pre-launch check: resolution, write set and authorization. */ diff --git a/ui/src/lib/api/http-client.ts b/ui/src/lib/api/http-client.ts index 7c908c7c95..52d835ca88 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -1,13 +1,15 @@ import createFetch from 'openapi-fetch'; import type { components, paths } from '$control/schema.d.ts'; import { ApiError, type ApiClient } from './client'; +import { RUN_WINDOW, runDetailOf, runRowPage, runTabCountsOf, toRunNode, toRuns } from './http-runs'; import type { ChangeSubscription, ChangedChannel, ChangedSignal, ConnectionState } from './signals'; import type { AuthMode, Bootstrap, Capability, Location, Scope } from './types'; +import type { RerunMode, Run, RunDetail, RunRowPage, RunRowQuery, RunTabCounts } from './types/operate'; type Wire = components['schemas']; /** The client methods the server has endpoints for; the shell hides pages that need others. */ -const SUPPORTED: ReadonlySet = new Set(['supports', 'bootstrap', 'changes']); +const SUPPORTED: ReadonlySet = new Set(['supports', 'bootstrap', 'changes', 'listRunRows', 'runTabCounts', 'getRun', 'getRunDetail', 'cancelRun', 'rerunRun']); /** A UI location is one server location and target; its id carries both. */ export function locationKey(locationId: string, target: string): string { @@ -162,8 +164,69 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt if (response.status === 401) onUnauthorized(); if (!data) throw new ApiError(errorText(error, response), response.status); return toBootstrap(data); + }, + listRunRows: async (scope: Scope, query: RunRowQuery = {}): Promise => runRowPage(await listedRuns(scope), scope.locationId, query), + runTabCounts: async (scope: Scope, query: RunRowQuery = {}): Promise => runTabCountsOf(await listedRuns(scope), scope.locationId, query), + getRun: async (scope: Scope, runId: string): Promise => (await findRun(scope, runId))?.run ?? null, + async getRunDetail(scope: Scope, runId: string): Promise { + const found = await findRun(scope, runId); + if (!found) return null; + const { run, wire, listed } = found; + const owners = [run, ...listed.filter((r) => r.parentRunId === run.id)]; + const nodes = (await Promise.all(owners.map(async (owner) => (await runNodes(scope, owner.id)).map((n) => toRunNode(n, owner))))).flat(); + return runDetailOf(run, wire, listed, nodes); + }, + async cancelRun(scope: Scope, runId: string): Promise { + const body = { run_id: runId, event_id: crypto.randomUUID(), reason: 'Cancelled from the web UI' }; + const { error, response } = await http.POST('/api/control/runs/cancel', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + if (response.status === 401) onUnauthorized(); + if (!response.ok) throw new ApiError(errorText(error, response), response.status); + return requireRun(scope, runId); + }, + async rerunRun(scope: Scope, runId: string, mode: Exclude): Promise { + const source = await requireRun(scope, runId); + const body = { run_id: crypto.randomUUID(), source_run_id: runId, from_failure: mode === 'from_failure', priority: source.priority, reason: 'Rerun from the web UI' }; + const { error, response } = await http.POST('/api/control/runs/rerun', { body, params: { header: { 'x-sqb-csrf': csrfToken(document.cookie) } } }); + if (response.status === 401) onUnauthorized(); + if (!response.ok) throw new ApiError(errorText(error, response), response.status); + return requireRun(scope, body.run_id); } }; + + /** The newest runs of the location, up to the server's cap. */ + async function listedRuns(scope: Scope) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/runs', { + params: { path: { location_id: locationId, target }, query: { limit: RUN_WINDOW } } + }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + + async function findRun(scope: Scope, runId: string) { + const wires = await listedRuns(scope); + const listed = toRuns(wires, scope.locationId); + const index = listed.findIndex((r) => r.id === runId); + return index < 0 ? null : { run: listed[index], wire: wires[index], listed }; + } + + async function requireRun(scope: Scope, runId: string): Promise { + const found = await findRun(scope, runId); + if (!found) throw new ApiError('The run is not in the recent runs of this location', 404); + return found.run; + } + + async function runNodes(scope: Scope, runId: string) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes', { + params: { path: { location_id: locationId, target, run_id: runId } } + }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + return new Proxy(implemented, { get(target, property, receiver) { if (property in target) return Reflect.get(target, property, receiver); diff --git a/ui/src/lib/api/http-runs.test.ts b/ui/src/lib/api/http-runs.test.ts new file mode 100644 index 0000000000..8e4f328486 --- /dev/null +++ b/ui/src/lib/api/http-runs.test.ts @@ -0,0 +1,124 @@ +import { describe, expect, it } from 'vitest'; +import type { components } from '$control/schema.d.ts'; +import { runDetailOf, runRowPage, runTabCountsOf, toRunNode, toRuns } from './http-runs'; + +type RunRecord = components['schemas']['RunRecord']; + +const LOCATION = '0191f0aa-orders:development'; +const PERSON_ID = '0191f0cc-0000-7000-8000-000000000001'; + +function wire(id: string, overrides: Partial = {}): RunRecord { + return { + run_id: id, + root_run_id: id, + kind: 'job', + origin: 'manual', + status: 'success', + job_key: 'orders_daily', + last_event_storage_id: 1, + location_id: '0191f0aa-0000-7000-8000-000000000000', + node_counts_json: {}, + priority: 0, + queued_at: '2026-11-06T10:00:00Z', + requested_by: `principal:${PERSON_ID}`, + requester_name: 'Ada Admin', + rewrite_history: false, + selection_json: ['model.orders'], + snapshot_id: '0191f0dd-0000-7000-8000-000000000000', + target: 'development', + ...overrides + }; +} + +describe('run mapping', () => { + it('shows a person by display name and a machine origin by its trigger', () => { + const [person, sensor, scheduler] = toRuns( + [ + wire('a'), + wire('b', { origin: 'sensor', requested_by: 'system:sensor', requester_name: null, trigger_name: 'new_inventory_file' }), + wire('c', { origin: 'schedule', requested_by: 'system:scheduler', requester_name: null }) + ], + LOCATION + ); + expect(person.requestedBy).toEqual({ kind: 'user', name: PERSON_ID, displayName: 'Ada Admin' }); + expect(sensor.requestedBy).toEqual({ kind: 'sensor', name: 'new_inventory_file', displayName: 'new_inventory_file' }); + expect(scheduler.requestedBy).toEqual({ kind: 'scheduler', name: 'system:scheduler', displayName: 'system:scheduler' }); + }); + + it('counts cancelled nodes as failed and counts each run’s direct children', () => { + const runs = toRuns( + [ + wire('group', { kind: 'group', node_counts_json: { success: 3, cancelled: 2, failed: 1, pending: 4 } }), + wire('child-1', { parent_run_id: 'group', root_run_id: 'group' }), + wire('child-2', { parent_run_id: 'group', root_run_id: 'group' }), + wire('grandchild', { parent_run_id: 'child-1', root_run_id: 'group' }) + ], + LOCATION + ); + expect(runs[0].nodeCounts).toEqual({ success: 3, warn: 0, failed: 3, skipped: 0, running: 0, pending: 4 }); + expect(runs.map((r) => r.childRunCount)).toEqual([2, 1, 0, 0]); + }); +}); + +describe('runs list', () => { + const listing = [ + wire('r1', { status: 'running', cancel_requested_at: '2026-11-06T10:05:00Z' }), + wire('r2', { status: 'cancelled', cancel_requested_at: '2026-11-06T09:05:00Z' }), + wire('r3', { status: 'queued' }), + wire('r4', { status: 'failed', node_counts_json: { success: 1, failed: 2, skipped: 3 } }), + wire('r5', { status: 'partial' }), + wire('r5-child', { status: 'queued', parent_run_id: 'r5', root_run_id: 'r5' }), + wire('r6', { status: 'success' }), + wire('r7', { status: 'cancelled', node_counts_json: { success: 1, failed: 1, cancelled: 2, skipped: 1 } }) + ]; + + it('pages one tab with an offset cursor and hides children unless asked', () => { + const first = runRowPage(listing, LOCATION, { tab: 'all', limit: 4 }); + expect(first.rows.map((r) => r.run.id)).toEqual(['r1', 'r2', 'r3', 'r4']); + expect(first.nextCursor).toBe('4'); + const second = runRowPage(listing, LOCATION, { tab: 'all', limit: 4, cursor: first.nextCursor }); + expect(second.rows.map((r) => r.run.id)).toEqual(['r5', 'r6', 'r7']); + expect(second.nextCursor).toBeNull(); + expect(runRowPage(listing, LOCATION, { tab: 'failed' }).rows.map((r) => r.run.id)).toEqual(['r4', 'r5']); + expect(runRowPage(listing, LOCATION, { parentRunId: 'r5' }).rows.map((r) => r.run.id)).toEqual(['r5-child']); + }); + + it('keeps a cancel request only while the run is active', () => { + const rows = runRowPage(listing, LOCATION, { tab: 'all' }).rows; + expect(rows.find((r) => r.run.id === 'r1')?.cancelRequestedAt).toBe('2026-11-06T10:05:00Z'); + expect(rows.find((r) => r.run.id === 'r2')?.cancelRequestedAt).toBeNull(); + expect(rows.find((r) => r.run.id === 'r4')?.rerunFromFailureCount).toBe(5); + expect(rows.find((r) => r.run.id === 'r7')?.rerunFromFailureCount).toBe(2); + }); + + it('counts active and queued runs under the filter, ignoring the tab', () => { + expect(runTabCountsOf(listing, LOCATION, { tab: 'failed' })).toEqual({ active: 2, queued: 1 }); + expect(runTabCountsOf(listing, LOCATION, { includeChildren: true })).toEqual({ active: 3, queued: 2 }); + expect(runTabCountsOf(listing, LOCATION, { filter: { requestedBy: ['ada admin'] }, includeChildren: true })).toEqual({ active: 3, queued: 2 }); + expect(runTabCountsOf(listing, LOCATION, { filter: { requestedBy: ['someone else'] } })).toEqual({ active: 0, queued: 0 }); + }); +}); + +describe('run detail', () => { + it('links parent, children and reruns, and reruns from failed and skipped nodes but not cancelled ones', () => { + const wires = [ + wire('parent', { kind: 'group' }), + wire('run', { parent_run_id: 'parent', root_run_id: 'parent', status: 'failed' }), + wire('child-late', { parent_run_id: 'run', root_run_id: 'parent', queued_at: '2026-11-06T10:02:00Z' }), + wire('child-early', { parent_run_id: 'run', root_run_id: 'parent', queued_at: '2026-11-06T10:01:00Z' }), + wire('retry', { rerun_of_run_id: 'run' }) + ]; + const runs = toRuns(wires, LOCATION); + const run = runs[1]; + const nodes = (['success', 'failed', 'skipped', 'cancelled', 'bogus'] as const).map((status, i) => + toRunNode({ node_key: `model.n${i}`, status, skipped_because_of: status === 'skipped' ? 'model.n1' : null } as components['schemas']['RunNodeRecord'], run) + ); + const detail = runDetailOf(run, wires[1], runs, nodes); + expect(detail.parent?.id).toBe('parent'); + expect(detail.children.map((r) => r.id)).toEqual(['child-early', 'child-late']); + expect(detail.reruns.map((r) => r.id)).toEqual(['retry']); + expect(detail.rerunFromFailureNodes).toEqual(['model.n1', 'model.n2']); + expect(nodes[4].status).toBe('not_run'); + expect(nodes[2].skippedBecauseOf).toEqual({ nodeKey: 'model.n1', runId: 'run', jobKey: 'orders_daily' }); + }); +}); diff --git a/ui/src/lib/api/http-runs.ts b/ui/src/lib/api/http-runs.ts new file mode 100644 index 0000000000..305ffa1848 --- /dev/null +++ b/ui/src/lib/api/http-runs.ts @@ -0,0 +1,196 @@ +import type { components } from '$control/schema.d.ts'; +import { isActiveRun } from '$lib/status'; +import type { Principal } from './types'; +import type { NodeCounts, Run, RunDetail, RunNode, RunNodeStatus, RunRow, RunRowPage, RunRowQuery, RunTabCounts } from './types/operate'; + +type Wire = components['schemas']; + +/** The server lists at most this many runs per request; tabs, filters and pages apply to them. */ +export const RUN_WINDOW = 1000; + +const PERSON = 'principal:'; + +/** Who requested a run: a person by display name and principal ID, otherwise the trigger or system actor. */ +export function requesterOf(wire: Wire['RunRecord']): Principal { + if (wire.requested_by.startsWith(PERSON)) { + const id = wire.requested_by.slice(PERSON.length); + return { kind: 'user', name: id, displayName: wire.requester_name ?? id }; + } + const actor = wire.requested_by; + const kind = actor === 'system:scheduler' ? 'scheduler' : actor === 'system:sensor' ? 'sensor' : 'system'; + const name = wire.trigger_name ?? actor; + return { kind, name, displayName: name }; +} + +function countsOf(wire: Wire['RunRecord']): NodeCounts { + const c = wire.node_counts_json; + const n = (key: string) => c[key] ?? 0; + return { + success: n('success'), + warn: n('warn'), + failed: n('failed') + n('cancelled'), + skipped: n('skipped'), + running: n('running'), + pending: n('pending') + }; +} + +export function toRun(wire: Wire['RunRecord'], locationId: string, childRunCount: number): Run { + return { + id: wire.run_id, + kind: wire.kind, + origin: wire.origin, + status: wire.status, + statusReason: wire.status_reason ?? null, + jobKey: wire.job_key ?? null, + parentRunId: wire.parent_run_id ?? null, + rootRunId: wire.root_run_id, + childRunCount, + triggerId: wire.trigger_id ?? null, + triggerName: wire.trigger_name ?? null, + tickId: wire.tick_id ?? null, + tickScheduledFor: null, + runKey: null, + rerunOfRunId: wire.rerun_of_run_id ?? null, + requestedBy: requesterOf(wire), + priority: wire.priority, + rewriteHistory: wire.rewrite_history, + locationId, + target: wire.target, + snapshotId: wire.snapshot_id, + gitRevision: null, + queuedAt: wire.queued_at, + launchedAt: wire.launched_at ?? null, + startedAt: wire.started_at ?? null, + endedAt: wire.ended_at ?? null, + nodeCounts: countsOf(wire), + cost: null + }; +} + +/** Runs keyed by id, each with its number of child runs in the listing. */ +export function toRuns(wires: Wire['RunRecord'][], locationId: string): Run[] { + const children = new Map(); + for (const w of wires) if (w.parent_run_id) children.set(w.parent_run_id, (children.get(w.parent_run_id) ?? 0) + 1); + return wires.map((w) => toRun(w, locationId, children.get(w.run_id) ?? 0)); +} + +function rowOf(run: Run, wire: Wire['RunRecord']): RunRow { + const c = run.nodeCounts; + const total = c.success + c.warn + c.failed + c.skipped + c.running + c.pending; + return { + run, + progress: { done: total - c.running - c.pending, total }, + waits: [], + lastContactAt: null, + lostContactLimitSeconds: 0, + cancelRequestedAt: isActiveRun(run.status) ? (wire.cancel_requested_at ?? null) : null, + selection: run.kind === 'adhoc' ? wire.selection_json : [], + rerunFromFailureCount: (wire.node_counts_json.failed ?? 0) + c.skipped + }; +} + +function inTab(run: Run, tab: RunRowQuery['tab']): boolean { + if (tab === 'active') return isActiveRun(run.status); + if (tab === 'queued') return run.status === 'queued'; + if (tab === 'failed') return run.status === 'failed' || run.status === 'partial' || run.status === 'presumed_failed'; + if (tab === 'cancelled') return run.status === 'cancelled'; + return true; +} + +/** Whether a run passes a runs-list query, apart from its tab. Node-key filters need nodes and match nothing here. */ +function matches(run: Run, wire: Wire['RunRecord'], query: RunRowQuery): boolean { + const f = query.filter ?? {}; + const lower = (xs?: string[]) => (xs?.length ? xs.map((x) => x.toLowerCase()) : null); + if (query.parentRunId) { + if (run.parentRunId !== query.parentRunId) return false; + } else if (!query.includeChildren && run.parentRunId) return false; + if (query.since && run.queuedAt < query.since) return false; + if (query.until && run.queuedAt >= query.until) return false; + if (f.kinds?.length && !f.kinds.includes(run.kind)) return false; + if (f.origins?.length && !f.origins.includes(run.origin)) return false; + if (f.statuses?.length && !f.statuses.includes(run.status)) return false; + const ids = lower(f.ids)?.map((id) => id.replace(/^run-/, '')); + if (ids && !ids.some((id) => run.id.toLowerCase().startsWith(id))) return false; + const jobs = lower(f.jobKeys); + if (jobs && !jobs.includes((run.jobKey ?? '').toLowerCase())) return false; + const trigger = (run.triggerName ?? '').toLowerCase(); + const schedules = lower(f.schedules); + if (schedules && !(run.origin === 'schedule' && schedules.includes(trigger))) return false; + const sensors = lower(f.sensors); + if (sensors && !(run.origin === 'sensor' && sensors.includes(trigger))) return false; + const by = lower(f.requestedBy); + if (by && !by.includes(run.requestedBy.name.toLowerCase()) && !by.includes(run.requestedBy.displayName.toLowerCase())) return false; + if (f.nodeKeys?.length && !f.nodeKeys.some((k) => wire.selection_json.includes(k))) return false; + const text = f.text?.trim().toLowerCase() ?? ''; + if (text && !`${run.id} ${run.jobKey ?? ''} ${run.triggerName ?? ''} ${run.requestedBy.name} ${run.requestedBy.displayName}`.toLowerCase().includes(text)) return false; + return true; +} + +/** One page of the listing under a runs-list query; the cursor is an offset into the newest runs. */ +export function runRowPage(wires: Wire['RunRecord'][], locationId: string, query: RunRowQuery): RunRowPage { + const runs = toRuns(wires, locationId); + const matching = runs.flatMap((run, i) => (matches(run, wires[i], query) && inTab(run, query.tab) ? [rowOf(run, wires[i])] : [])); + const offset = Number(query.cursor ?? 0) || 0; + const limit = query.limit ?? 50; + return { + rows: matching.slice(offset, offset + limit), + nextCursor: offset + limit < matching.length ? String(offset + limit) : null + }; +} + +export function runTabCountsOf(wires: Wire['RunRecord'][], locationId: string, query: RunRowQuery): RunTabCounts { + const runs = toRuns(wires, locationId).filter((run, i) => matches(run, wires[i], query)); + return { + active: runs.filter((r) => isActiveRun(r.status)).length, + queued: runs.filter((r) => r.status === 'queued').length + }; +} + +function nodeStatus(status: string): RunNodeStatus { + return (['pending', 'running', 'success', 'warn', 'failed', 'skipped', 'cancelled'] as const).find((s) => s === status) ?? 'not_run'; +} + +export function toRunNode(wire: Wire['RunNodeRecord'], run: Run): RunNode { + return { + nodeKey: wire.node_key, + kind: null, + runId: run.id, + jobKey: run.jobKey, + status: nodeStatus(wire.status), + attempt: 1, + startedAt: wire.started_at ?? null, + endedAt: wire.ended_at ?? null, + rowsAffected: wire.rows_affected ?? null, + errorCode: wire.error_code ?? null, + errorMessage: null, + errorLocation: null, + errorDetail: null, + skippedBecauseOf: wire.skipped_because_of ? { nodeKey: wire.skipped_because_of, runId: run.id, jobKey: run.jobKey } : null, + critical: false + }; +} + +/** + * A run's detail from the listing and the nodes of the run and its child runs; parts with no server source stay empty. + * Rerun from failure covers failed and skipped nodes only, as the server does; cancelled nodes need a full rerun. + */ +export function runDetailOf(run: Run, wire: Wire['RunRecord'], listed: Run[], nodes: RunNode[]): RunDetail { + const byId = new Map(listed.map((r) => [r.id, r])); + return { + run, + parent: run.parentRunId ? (byId.get(run.parentRunId) ?? null) : null, + children: listed.filter((r) => r.parentRunId === run.id).sort((a, b) => a.queuedAt.localeCompare(b.queuedAt)), + nodes, + edges: [], + invocations: [], + liveness: null, + cancel: null, + provenance: null, + selection: wire.selection_json, + waits: [], + rerunFromFailureNodes: nodes.filter((n) => n.status === 'failed' || n.status === 'skipped').map((n) => n.nodeKey), + rerunOf: run.rerunOfRunId ? (byId.get(run.rerunOfRunId) ?? null) : null, + reruns: listed.filter((r) => r.rerunOfRunId === run.id) + }; +} diff --git a/ui/src/lib/api/mock/mock-client.ts b/ui/src/lib/api/mock/mock-client.ts index dde29a1485..0e632aba05 100644 --- a/ui/src/lib/api/mock/mock-client.ts +++ b/ui/src/lib/api/mock/mock-client.ts @@ -229,9 +229,14 @@ export class MockApiClient implements ApiClient { return this.#operate.cancelRun(scope, runId); } - async rerunRun(scope: Scope, runId: string, mode: RerunMode, nodes?: string[]): Promise { + async rerunRun(scope: Scope, runId: string, mode: Exclude): Promise { this.#location(scope); - return this.#operate.rerunRun(scope, runId, mode, nodes); + return this.#operate.rerunRun(scope, runId, mode); + } + + async rerunSelectedNodes(scope: Scope, runId: string, nodes: string[]): Promise { + this.#location(scope); + return this.#operate.rerunRun(scope, runId, 'selected', nodes); } getStatementSql(...args: Parameters) { diff --git a/ui/src/lib/api/mock/operate-store.ts b/ui/src/lib/api/mock/operate-store.ts index 89037a9737..9765fa7a3a 100644 --- a/ui/src/lib/api/mock/operate-store.ts +++ b/ui/src/lib/api/mock/operate-store.ts @@ -270,7 +270,7 @@ export class MockOperate { provenance: { snapshotId: rec.run.snapshotId, snapshotGeneration: rec.script.snapshotGeneration, - gitRevision: rec.run.gitRevision, + gitRevision: rec.run.gitRevision ?? '', gitDirty: false, imageId: rec.script.imageId, sqbVersion: rec.script.snapshotGeneration === SNAPSHOT_GENERATION ? '0.185.0' : '0.184.2', @@ -278,6 +278,7 @@ export class MockOperate { exclude: rec.script.exclude, command: rec.script.nodes.length ? commandOf(rec) : `group ${rec.run.jobKey}: ${this.#children(rec).length} child runs` }, + selection: rec.script.select, waits: rec.run.status === 'queued' ? this.#queue().get(rec.run.id) ?? [] : [], rerunFromFailureNodes: nodes.filter((n) => n.status === 'failed' || n.status === 'skipped').map((n) => n.nodeKey), rerunOf: rec.run.rerunOfRunId ? ((r) => (r ? this.#view(r, scope) : null))(this.#byId(rec.run.rerunOfRunId)) : null, diff --git a/ui/src/lib/api/types/operate.ts b/ui/src/lib/api/types/operate.ts index 32f616ccd6..40bbd8a81a 100644 --- a/ui/src/lib/api/types/operate.ts +++ b/ui/src/lib/api/types/operate.ts @@ -27,7 +27,7 @@ export const RUN_KINDS = ['job', 'group', 'adhoc'] as const; export type RunKind = (typeof RUN_KINDS)[number]; /** Why the run was requested. Group children inherit their parent's origin. */ -export const RUN_ORIGINS = ['manual', 'schedule', 'sensor', 'rerun'] as const; +export const RUN_ORIGINS = ['manual', 'schedule', 'sensor', 'rerun', 'backfill'] as const; export type RunOrigin = (typeof RUN_ORIGINS)[number]; export type NodeCounts = { @@ -74,7 +74,8 @@ export type Run = { locationId: string; target: string; snapshotId: string; - gitRevision: string; + /** Null where the server does not report the commit. */ + gitRevision: string | null; queuedAt: Iso; launchedAt: Iso | null; startedAt: Iso | null; @@ -197,7 +198,8 @@ export type SkippedBecauseOf = { export type RunNode = { nodeKey: string; - kind: NodeKind; + /** Null where the server does not report the node kind. */ + kind: NodeKind | null; /** The (child) run that executed this node. */ runId: string; jobKey: string | null; @@ -359,7 +361,10 @@ export type RunDetail = { invocations: RunInvocation[]; liveness: RunLiveness | null; cancel: CancelProgress | null; - provenance: RunProvenance; + /** Null where the server does not report how the run was built. */ + provenance: RunProvenance | null; + /** The frozen selection the run was authorized for. */ + selection: string[]; /** Why the run is still queued; empty once claimed. */ waits: QueueWait[]; /** Nodes a rerun from failure would run: the failed ones plus exactly those skipped. */ diff --git a/ui/src/lib/operate/labels.ts b/ui/src/lib/operate/labels.ts index 414f2b9f7c..8715ef3349 100644 --- a/ui/src/lib/operate/labels.ts +++ b/ui/src/lib/operate/labels.ts @@ -100,9 +100,10 @@ export const onFailurePlain: Record = { export type WaitPart = { text: string; href?: string; code?: boolean }; -/** `run-4f2a` → `4f2a`: the short form shown in lists; the full ID stays in links and copies. */ +/** `run-4f2a` → `4f2a`, a UUID → its first 8 digits: the short form shown in lists; the full ID stays in links and copies. */ export function shortRunId(id: string): string { - return id.replace(/^run-/, ''); + const short = id.replace(/^run-/, ''); + return /^[0-9a-f]{8}-[0-9a-f]{4}-/i.test(short) ? short.slice(0, 8) : short; } /** A concurrency group's slots in words, e.g. `warehouse_heavy` → `warehouse heavy`, `target:prod` → `prod target`. */ diff --git a/ui/src/lib/status.ts b/ui/src/lib/status.ts index 76e24d8449..dc7db40e72 100644 --- a/ui/src/lib/status.ts +++ b/ui/src/lib/status.ts @@ -141,7 +141,8 @@ export const runOriginLabel: Record = { manual: 'Manual', schedule: 'Schedule', sensor: 'Sensor', - rerun: 'Rerun' + rerun: 'Rerun', + backfill: 'Backfill' }; /** One-line "what · why" label, e.g. "Group · Schedule". */ diff --git a/ui/src/routes/runs/+page.svelte b/ui/src/routes/runs/+page.svelte index ca212214a2..16060abb69 100644 --- a/ui/src/routes/runs/+page.svelte +++ b/ui/src/routes/runs/+page.svelte @@ -1,5 +1,6 @@ - (panelOpen = true)}>▷ New run + {#if app.client.supports('launchAdhocRun')} + (panelOpen = true)}>▷ New run + {/if} -{#if p.view === 'timeline'} +{#if shown === 'timeline'}
-{:else if p.view === 'queue'} +{:else if shown === 'queue'} {:else} {/if} - +{#if app.client.supports('launchAdhocRun')}{/if}