feat(ui): wire the runs pages to the live control API

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:<id>` 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: {

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<RunRecord>), 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
 }
 

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<String>,
+    pub kind: RunKind,
+    pub origin: RunOrigin,
+    pub parent_run_id: Option<Uuid>,
+    pub root_run_id: Uuid,
+    pub trigger_id: Option<Uuid>,
+    pub trigger_name: Option<String>,
+    pub tick_id: Option<Uuid>,
+    pub rerun_of_run_id: Option<Uuid>,
+    /// The stored actor: `principal:<id>` 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<String>,
+    pub priority: i64,
+    pub rewrite_history: bool,
     pub status: RunStatus,
+    pub status_reason: Option<String>,
     pub queued_at: DateTime<Utc>,
     pub launched_at: Option<DateTime<Utc>>,
     pub started_at: Option<DateTime<Utc>>,
     pub ended_at: Option<DateTime<Utc>>,
+    pub cancel_requested_at: Option<DateTime<Utc>>,
     pub selection_json: BTreeSet<String>,
     pub node_counts_json: BTreeMap<String, i64>,
     pub last_event_storage_id: i64,

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),

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

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;

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<Run>;
 	/** `nodes` is required for `selected` and ignored otherwise. */
-	rerunRun(scope: Scope, runId: string, mode: RerunMode, nodes?: string[]): Promise<Run>;
+	rerunRun(scope: Scope, runId: string, mode: Exclude<RerunMode, 'selected'>): Promise<Run>;
+	/** Reruns exactly `nodes` of the run. */
+	rerunSelectedNodes(scope: Scope, runId: string, nodes: string[]): Promise<Run>;
 
 	listSnapshotNodes(scope: Scope): Promise<SnapshotNode[]>;
 	/** The server's pre-launch check: resolution, write set and authorization. */

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<string> = new Set<keyof ApiClient>(['supports', 'bootstrap', 'changes']);
+const SUPPORTED: ReadonlySet<string> = new Set<keyof ApiClient>(['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> => runRowPage(await listedRuns(scope), scope.locationId, query),
+		runTabCounts: async (scope: Scope, query: RunRowQuery = {}): Promise<RunTabCounts> => runTabCountsOf(await listedRuns(scope), scope.locationId, query),
+		getRun: async (scope: Scope, runId: string): Promise<Run | null> => (await findRun(scope, runId))?.run ?? null,
+		async getRunDetail(scope: Scope, runId: string): Promise<RunDetail | null> {
+			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<Run> {
+			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<RerunMode, 'selected'>): Promise<Run> {
+			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<Run> {
+		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);