From 96a0f8fb888b27cfb4c1c3101eb457123126b82b Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:33:58 +0000 Subject: [PATCH 1/3] feat(orchestration): let people list a target's published jobs and groups GET /api/control/locations/{location_id}/targets/{target}/jobs returns each current job and group with its kind, description, selection, concurrency groups and snapshot, under the same authorization as the trigger listing. Settings stay out of person reads. Validated: api and declarations tests on SQLite and PostgreSQL, including the session role matrix and the denial matrix. --- clients/typescript/src/schema.d.ts | 64 +++++++++++++++++++ .../authentication/password_sessions.rs | 4 ++ .../api/_helpers/resources/person_reads.rs | 26 +++++++- .../sqlbuild-orchestration/src/api/models.rs | 1 + .../src/declarations/_helpers/reads.rs | 29 ++++++++- .../src/declarations/main/list_person_jobs.rs | 29 +++++++++ .../src/declarations/main/mod.rs | 1 + .../src/declarations/models.rs | 21 ++++++ .../tests/api/test_denials.rs | 3 +- .../tests/api/test_person_sessions.rs | 33 ++++++++++ fensu.toml | 2 +- 11 files changed, 207 insertions(+), 6 deletions(-) create mode 100644 crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 7c24125d11..8304565a96 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -354,6 +354,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/jobs": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["listPersonJobs"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/maintenance": { parameters: { query?: never; @@ -1134,6 +1150,17 @@ export interface components { failed: components["schemas"]["Table"][]; failure?: string | null; }; + /** @description A job or group of the current published snapshot. Settings are left out of person reads. */ + JobRecord: { + concurrency_groups: string[]; + description: string; + job_key: string; + /** @description `job` or `group`; a group's selection is the union of its member jobs' selections. */ + kind: components["schemas"]["RunKind"]; + selection: string[]; + /** Format: uuid */ + snapshot_id: string; + }; /** @description A trusted operator names a published job; execution inputs are never supplied by the caller. */ JobRunRequest: { job_key: string; @@ -2885,6 +2912,43 @@ export interface operations { }; }; }; + listPersonJobs: { + parameters: { + query?: never; + header?: never; + path: { + location_id: string; + target: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["JobRecord"][]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonMaintenance: { parameters: { query?: { diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs index 35f6359639..be80c63098 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -101,6 +101,10 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes", get(crate::api::_helpers::resources::person_reads::run_nodes), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/jobs", + get(crate::api::_helpers::resources::person_reads::jobs), + ) .route( "/api/control/locations/{location_id}/targets/{target}/triggers", get(crate::api::_helpers::resources::person_reads::triggers), diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs index cd4392dea9..3a7d8f6d34 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_reads.rs @@ -1,4 +1,4 @@ -//! Person reads of run nodes, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden. +//! Person reads of run nodes, jobs, triggers and ticks; refusal precedes every lookup, so unknown objects look forbidden. use axum::extract::{Path, Query, State}; use axum::http::StatusCode; @@ -8,8 +8,9 @@ use uuid::Uuid; use crate::api::errors::ApiError; use crate::api::models::ReadQuery; use crate::credentials::models::{AuditContext, AuthenticatedPerson, DefaultRole}; +use crate::declarations::main::list_person_jobs::list_person_jobs; use crate::declarations::main::list_person_triggers::list_person_triggers; -use crate::declarations::models::TriggerRecord; +use crate::declarations::models::{JobRecord, TriggerRecord}; use crate::runs::main::list_person_run_nodes::list_person_run_nodes; use crate::runs::models::RunNodeRecord; use crate::scheduler::main::list_person_ticks::list_person_ticks; @@ -39,6 +40,27 @@ pub(crate) async fn run_nodes( .ok_or_else(|| not_found("run not found")) } +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/jobs", operation_id = "listPersonJobs", + params(("location_id" = Uuid, Path), ("target" = String, Path)), + responses((status = 200, body = Vec), (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn jobs( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + Path((location_id, target)): Path<(Uuid, String)>, + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result>, ApiError> { + Ok(Json( + list_person_jobs( + &store, + (&context, principal), + (location_id, &target), + default_role, + ) + .await?, + )) +} + #[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/triggers", operation_id = "listPersonTriggers", params(("location_id" = Uuid, Path), ("target" = String, Path)), responses((status = 200, body = Vec), (status = 401, description = "Session rejected"), diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index 17dc9477f4..d52b2bac13 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -21,6 +21,7 @@ use crate::api::_helpers::resources::{ person_operations::change_trigger, person_operations::recent_runs, person_reads::run_nodes, + person_reads::jobs, person_reads::triggers, person_reads::ticks, person_operations::catalog, diff --git a/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs b/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs index a69248fd01..00f3bf5150 100644 --- a/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs +++ b/crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs @@ -1,8 +1,8 @@ -//! Trigger reads inside a caller's transaction, so authorization and results share one view. +//! Job and trigger reads inside a caller's transaction, so authorization and results share one view. use uuid::Uuid; -use crate::declarations::models::TriggerRecord; +use crate::declarations::models::{JobRecord, StoredJob, TriggerRecord}; use crate::store::errors::ControlError; use crate::store::main::query_records::query_records; use crate::store::models::{ControlTransaction, Parameter}; @@ -17,3 +17,28 @@ pub(crate) async fn triggers_in( WHERE t.location_id=$1 AND t.target=$2 AND t.removed_in_snapshot_id IS NULL ORDER BY t.name", vec![Parameter::Id(location_id), Parameter::Text(target.into())]).await } + +pub(crate) async fn jobs_in( + transaction: ControlTransaction, + (location_id, target): (Uuid, &str), +) -> Result<(ControlTransaction, Vec), ControlError> { + let (transaction, stored): (_, Vec) = query_records( + transaction, + "SELECT job_key, kind, snapshot_id, definition_json FROM sqb_jobs \ + WHERE location_id=$1 AND target=$2 ORDER BY job_key", + vec![Parameter::Id(location_id), Parameter::Text(target.into())], + ) + .await?; + let jobs: Vec = stored + .into_iter() + .map(|job| JobRecord { + job_key: job.job_key, + kind: job.kind, + description: job.definition_json.description, + selection: job.definition_json.selection, + concurrency_groups: job.definition_json.concurrency_groups, + snapshot_id: job.snapshot_id, + }) + .collect(); + Ok((transaction, jobs)) +} diff --git a/crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs b/crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs new file mode 100644 index 0000000000..8675f3e0d0 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs @@ -0,0 +1,29 @@ +//! A person lists jobs only with view on every current node of the target, like triggers. + +use uuid::Uuid; + +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::declarations::_helpers::reads::jobs_in; +use crate::declarations::models::JobRecord; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit}; +use crate::store::models::{ControlStore, TransactionMode}; + +pub async fn list_person_jobs( + store: &ControlStore, + (context, credentials): (&AuditContext, Uuid), + (location_id, target): (Uuid, &str), + default_role: DefaultRole, +) -> Result, ControlError> { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + (context, credentials), + (location_id, target), + default_role, + ) + .await?; + let (transaction, jobs) = jobs_in(transaction, (location_id, target)).await?; + commit(transaction).await?; + Ok(jobs) +} diff --git a/crates/sqlbuild-orchestration/src/declarations/main/mod.rs b/crates/sqlbuild-orchestration/src/declarations/main/mod.rs index 1a14985c37..0729a456c1 100644 --- a/crates/sqlbuild-orchestration/src/declarations/main/mod.rs +++ b/crates/sqlbuild-orchestration/src/declarations/main/mod.rs @@ -2,6 +2,7 @@ pub mod change_person_trigger; pub mod change_trigger; pub(crate) mod frozen_definitions; pub mod group_dependencies; +pub mod list_person_jobs; pub mod list_person_triggers; pub mod list_trigger_events; pub mod list_triggers; diff --git a/crates/sqlbuild-orchestration/src/declarations/models.rs b/crates/sqlbuild-orchestration/src/declarations/models.rs index afdcab28d0..726975e856 100644 --- a/crates/sqlbuild-orchestration/src/declarations/models.rs +++ b/crates/sqlbuild-orchestration/src/declarations/models.rs @@ -87,6 +87,27 @@ pub enum TriggerDeclaration { Sensor(SensorDefinition), } +/// A job or group of the current published snapshot. Settings are left out of person reads. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct JobRecord { + pub job_key: String, + /// `job` or `group`; a group's selection is the union of its member jobs' selections. + pub kind: crate::runs::models::RunKind, + pub description: String, + pub selection: BTreeSet, + pub concurrency_groups: BTreeSet, + pub snapshot_id: uuid::Uuid, +} + +/// A `sqb_jobs` row as stored. +#[derive(Clone, Debug, Deserialize)] +pub(crate) struct StoredJob { + pub job_key: String, + pub kind: crate::runs::models::RunKind, + pub snapshot_id: uuid::Uuid, + pub definition_json: JobDefinition, +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, utoipa::ToSchema)] pub struct TriggerRecord { pub trigger_id: uuid::Uuid, diff --git a/crates/sqlbuild-orchestration/tests/api/test_denials.rs b/crates/sqlbuild-orchestration/tests/api/test_denials.rs index 7f173d7916..439f159c7a 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_denials.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_denials.rs @@ -20,8 +20,9 @@ use crate::store::tests::store_helpers::{close_test_store, open_test_store, run} use crate::store::tests::store_test_types::Backend; /// Location-scoped reads and the action each needs. -const READS: [(&str, &str); 10] = [ +const READS: [(&str, &str); 11] = [ ("runs", "view"), + ("jobs", "view"), ("triggers", "view"), ("changes", "view"), ("catalog?schema=orders", "view"), diff --git a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs index 6984b0a561..9753574753 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_person_sessions.rs @@ -93,6 +93,12 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ let mut input = snapshot(location); input.documents.get_mut("triggers").expect("triggers")["health_orders"]["when"] = json!({"kind":"custom", "function":"orders_ready", "code_hash":"orders_v1"}); + input.documents.get_mut("jobs").expect("jobs")["orders"]["concurrency_groups"] = + json!(["warehouse_heavy"]); + input.documents.insert( + "groups".into(), + json!({"nightly": {"description": "Nightly orders", "selection": ["orders"], "jobs": {"orders": []}}}), + ); let sql_input: Value = serde_json::from_str( &std::fs::read_to_string( std::path::Path::new(env!("CARGO_MANIFEST_DIR")) @@ -279,6 +285,7 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ assert!(!page.to_string().contains("select 1 as id")); let known_reads = [ format!("{metadata}/{}/nodes", seed_run.run_id), + format!("/api/control/locations/{location}/targets/development/jobs"), format!("/api/control/locations/{location}/targets/development/triggers"), format!( "/api/control/locations/{location}/targets/development/triggers/{}/ticks", @@ -312,6 +319,32 @@ fn given_browser_session_roles_when_work_is_requested_then_scope_revocation_and_ "{read_path}" ); } + let jobs: Value = request( + &provider.client, + session, + ( + Method::GET, + &format!("/api/control/locations/{location}/targets/development/jobs"), + ), + (&Value::Null, &session.csrf), + ) + .await + .json() + .await + .expect("jobs response"); + let listed_jobs: Vec = + jobs.as_array().into_iter().flatten().cloned().collect(); + let expected_jobs = [ + json!({"job_key": "nightly", "kind": "group", "description": "Nightly orders", + "selection": ["orders"], "concurrency_groups": [], "snapshot_id": published.snapshot_id}), + json!({"job_key": "orders", "kind": "job", "description": "Refresh orders", + "selection": ["orders"], "concurrency_groups": ["warehouse_heavy"], "snapshot_id": published.snapshot_id}), + ]; + assert_eq!( + listed_jobs, + expected_jobs[..expected_jobs.len() * usize::from(jobs.is_array())], + "jobs list declared metadata and never settings" + ); for read_path in &unknown_reads { let missing = request( &provider.client, diff --git a/fensu.toml b/fensu.toml index fbabf3ab4c..c66c31072f 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1381,7 +1381,7 @@ paths = ["crates/sqlbuild-orchestration/src/api/_helpers/resources/administrator reason = "Each function is one utoipa-annotated axum route: OpenAPI needs one annotated handler per operation, and each body only checks the caller's scope and calls one store operation. Scope checks and store work live in their shared owners." [[dupes.allowlist]] -paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_person_ticks.rs", "crates/sqlbuild-orchestration/src/store/main/read_records.rs", "crates/sqlbuild-orchestration/src/platform/main/location_work.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] +paths = ["crates/sqlbuild-orchestration/src/authorization/main/read_security_audit.rs", "crates/sqlbuild-orchestration/src/credentials/_helpers/active_session.rs", "crates/sqlbuild-orchestration/src/declarations/_helpers/reads.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_jobs.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_person_triggers.rs", "crates/sqlbuild-orchestration/src/declarations/main/list_triggers.rs", "crates/sqlbuild-orchestration/src/platform/main/read_location_work.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_runs.rs", "crates/sqlbuild-orchestration/src/runs/main/list_person_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_run_nodes.rs", "crates/sqlbuild-orchestration/src/runs/main/list_runs.rs", "crates/sqlbuild-orchestration/src/runs/_helpers/lifecycle/reads.rs", "crates/sqlbuild-orchestration/src/warehouse_history/_helpers/checks.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_person_ticks.rs", "crates/sqlbuild-orchestration/src/store/main/read_records.rs", "crates/sqlbuild-orchestration/src/platform/main/location_work.rs", "crates/sqlbuild-orchestration/src/scheduler/main/list_ticks.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/attempts.rs", "crates/sqlbuild-orchestration/src/sensors/_helpers/facts.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_current_record.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_documents.rs", "crates/sqlbuild-orchestration/src/snapshots/main/read_snapshot_nodes.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_catalog.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_check.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_checks.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_maintenance.rs", "crates/sqlbuild-orchestration/src/warehouse_history/main/read_person_microbatch_coverage.rs", "crates/sqlbuild-orchestration/src/warehouse_metadata/main/read_person_statistics.rs"] reason = "Each is one control-store read with its own SQL and record type. Transactions, person authorization and decoding go through the shared begin_authorized, query_records, read_records and query_bodies helpers; what remains alike is the call shape." [[dupes.allowlist]] From 147d4f339aadaf06c7a54fd46f467eb61a588575 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:44:03 +0000 Subject: [PATCH 2/3] feat(ui): wire the jobs and triggers pages to the control API --- ui/src/lib/api/client.ts | 2 + ui/src/lib/api/http-client.ts | 115 +++++++++++++- ui/src/lib/api/http-jobs.test.ts | 65 ++++++++ ui/src/lib/api/http-jobs.ts | 144 ++++++++++++++++++ ui/src/lib/api/http-runs.ts | 19 ++- ui/src/lib/api/mock/mock-client.ts | 5 + ui/src/lib/api/mock/operate-store.ts | 3 +- ui/src/lib/api/mock/operate.ts | 8 +- ui/src/lib/api/types/operate.ts | 42 ++--- ui/src/routes/jobs/+page.svelte | 36 +++-- ui/src/routes/jobs/[name]/group-order.svelte | 4 +- ui/src/routes/jobs/[name]/job-view.svelte | 67 ++++---- ui/src/routes/jobs/[name]/launch-form.svelte | 13 +- .../routes/triggers/[id]/sensor-panel.svelte | 14 +- ui/src/routes/triggers/[id]/tick-log.svelte | 8 +- .../routes/triggers/[id]/trigger-view.svelte | 18 ++- 16 files changed, 475 insertions(+), 88 deletions(-) create mode 100644 ui/src/lib/api/http-jobs.test.ts create mode 100644 ui/src/lib/api/http-jobs.ts diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index fd900528bf..11d6581757 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -109,6 +109,8 @@ export interface ApiClient extends PlatformApi { listJobs(scope: Scope): Promise; getJob(scope: Scope, key: string): Promise; launchJob(scope: Scope, key: string, options?: JobLaunchOptions): Promise; + /** Launches with `--full-refresh`, rebuilding the job's tables from scratch. */ + launchJobFromScratch(scope: Scope, key: string, options?: JobLaunchOptions): Promise; listTriggers(scope: Scope): Promise; getTrigger(scope: Scope, id: string): Promise; diff --git a/ui/src/lib/api/http-client.ts b/ui/src/lib/api/http-client.ts index 52d835ca88..65927ad37f 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -1,15 +1,35 @@ 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 { jobDetailOf, toJob, toTick, toTrigger, triggerDetailOf } from './http-jobs'; +import { RUN_WINDOW, runDetailOf, runRowPage, runsMatching, 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'; +import type { Job, JobDetail, RerunMode, Run, RunDetail, RunListQuery, RunRowPage, RunRowQuery, RunTabCounts, Trigger, TriggerDetail, TriggerState } 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', 'listRunRows', 'runTabCounts', 'getRun', 'getRunDetail', 'cancelRun', 'rerunRun']); +const SUPPORTED: ReadonlySet = new Set([ + 'supports', + 'bootstrap', + 'changes', + 'listRunRows', + 'runTabCounts', + 'listRuns', + 'getRun', + 'getRunDetail', + 'cancelRun', + 'rerunRun', + 'listJobs', + 'getJob', + 'launchJob', + 'listTriggers', + 'getTrigger', + 'setTriggerStates', + 'resetTriggerToDefault', + 'resetSensorCursor' +]); /** A UI location is one server location and target; its id carries both. */ export function locationKey(locationId: string, target: string): string { @@ -190,9 +210,98 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt if (response.status === 401) onUnauthorized(); if (!response.ok) throw new ApiError(errorText(error, response), response.status); return requireRun(scope, body.run_id); + }, + listRuns: async (scope: Scope, query: RunListQuery = {}): Promise => runsMatching(toRuns(await listedRuns(scope), scope.locationId), query), + async listJobs(scope: Scope): Promise { + const [jobs, triggers, runs] = await Promise.all([listedJobs(scope), triggersOf(scope), recentRuns(scope)]); + return jobs.map((job) => toJob(job, triggers, runs)); + }, + async getJob(scope: Scope, key: string): Promise { + const [jobs, triggers, runs] = await Promise.all([listedJobs(scope), triggersOf(scope), recentRuns(scope)]); + const wire = jobs.find((job) => job.job_key === key); + return wire ? jobDetailOf(toJob(wire, triggers, runs), wire) : null; + }, + async launchJob(scope: Scope, key: string): Promise { + const { locationId, target } = parseLocationKey(scope.locationId); + const body = { run_id: crypto.randomUUID(), location_id: locationId, target, job_key: key, priority: 0, reason: 'Started from the web UI' }; + const { error, response } = await http.POST('/api/control/runs', { 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); + }, + listTriggers: (scope: Scope): Promise => triggersOf(scope), + async getTrigger(scope: Scope, id: string): Promise { + const [wires, runs] = await Promise.all([listedTriggers(scope), recentRuns(scope)]); + const wire = wires.find((t) => t.trigger_id === id); + if (!wire) return null; + const ticks = (await ticksOf(scope, id, 100)).map(toTick); + return triggerDetailOf(toTrigger(wire, runs, ticks[0] ?? null), wire, ticks); + }, + async setTriggerStates(scope: Scope, ids: string[], state: TriggerState, reason?: string | null): Promise { + for (const id of ids) await changeTrigger(id, { kind: state === 'running' ? 'enable' : 'pause' }, reason); + return (await triggersOf(scope)).filter((t) => ids.includes(t.id)); + }, + async resetTriggerToDefault(scope: Scope, id: string): Promise { + await changeTrigger(id, { kind: 'reset_default' }); + return requireTrigger(scope, id); + }, + async resetSensorCursor(scope: Scope, id: string, reason?: string | null): Promise { + await changeTrigger(id, { kind: 'reset_cursor' }, reason); + const detail = await implemented.getTrigger(scope, id); + if (!detail) throw new ApiError('The trigger is no longer published', 404); + return detail; } }; + async function listedJobs(scope: Scope) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/jobs', { params: { path: { location_id: locationId, target } } }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + + async function listedTriggers(scope: Scope) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/triggers', { params: { path: { location_id: locationId, target } } }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + + async function ticksOf(scope: Scope, triggerId: string, limit: number) { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/triggers/{trigger_id}/ticks', { + params: { path: { location_id: locationId, target, trigger_id: triggerId }, query: { limit } } + }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return data; + } + + const recentRuns = async (scope: Scope) => toRuns(await listedRuns(scope), scope.locationId); + + /** Triggers with their latest tick and the newest listed run each started. */ + async function triggersOf(scope: Scope): Promise { + const [wires, runs] = await Promise.all([listedTriggers(scope), recentRuns(scope)]); + const latest = await Promise.all(wires.map(async (t) => (await ticksOf(scope, t.trigger_id, 1)).map(toTick)[0] ?? null)); + return wires.map((t, i) => toTrigger(t, runs, latest[i])); + } + + async function requireTrigger(scope: Scope, id: string): Promise { + const trigger = (await triggersOf(scope)).find((t) => t.id === id); + if (!trigger) throw new ApiError('The trigger is no longer published', 404); + return trigger; + } + + /** Records an operator change; the server takes the actor from the session and needs a reason. */ + async function changeTrigger(triggerId: string, change: Wire['TriggerChange'], reason?: string | null) { + const body = { trigger_id: triggerId, event_id: crypto.randomUUID(), actor: '', reason: reason?.trim() || 'Changed from the web UI', change }; + const { error, response } = await http.POST('/api/control/triggers', { 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); + } + /** The newest runs of the location, up to the server's cap. */ async function listedRuns(scope: Scope) { const { locationId, target } = parseLocationKey(scope.locationId); diff --git a/ui/src/lib/api/http-jobs.test.ts b/ui/src/lib/api/http-jobs.test.ts new file mode 100644 index 0000000000..032968628f --- /dev/null +++ b/ui/src/lib/api/http-jobs.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it } from 'vitest'; +import type { components } from '$control/schema.d.ts'; +import { cadenceOf, jobDetailOf, toJob, toTick, toTrigger, triggerDetailOf, triggerKindOf } from './http-jobs'; +import type { Run } from './types/operate'; + +type W = components['schemas']; + +const SNAPSHOT = '0191f0dd-0000-7000-8000-000000000000'; + +function trigger(id: string, name: string, jobKey: string, definition: W['TriggerDeclaration'], status: W['TriggerStatus'] = 'running'): W['TriggerRecord'] { + return { trigger_id: id, name, job_key: jobKey, definition_json: definition, status, last_tick_for: null, next_due_at: '2026-11-07T06:00:00Z' }; +} + +const daily = trigger('t-daily', 'orders_daily', 'orders', { description: 'Daily orders', job_key: 'orders', schedule: { kind: 'cron', expression: '0 6 * * *', timezone: 'UTC' }, default_status: 'paused' }); +const onInventory = trigger('t-inv', 'inventory_landed', 'inventory', { + description: 'New stock feed', + job_key: 'inventory', + minimum_interval_seconds: 60, + when: { kind: 'source_new_data', selection: ['source:stock_feed'] } +}); + +function run(id: string, jobKey: string, triggerId: string | null): Run { + return { id, jobKey, triggerId, status: 'success', startedAt: '2026-11-07T01:00:00Z', endedAt: '2026-11-07T01:01:00Z' } as Run; +} + +describe('trigger mapping', () => { + it('classifies schedules and sensors and describes their cadence', () => { + expect(triggerKindOf(daily.definition_json)).toBe('cron_schedule'); + expect(triggerKindOf(onInventory.definition_json)).toBe('node_sensor'); + expect(triggerKindOf({ ...onInventory.definition_json, when: { kind: 'run_event', job_key: 'orders', kinds: ['succeeded' as never] } } as W['SensorDefinition'])).toBe('job_sensor'); + expect(cadenceOf(daily.definition_json)).toBe('0 6 * * * (UTC)'); + expect(cadenceOf(onInventory.definition_json)).toBe('when source:stock_feed has new data'); + }); + + it('keeps the code default apart from the effective state and links the newest run it started', () => { + const runs = [run('r-new', 'orders', 't-daily'), run('r-manual', 'orders', null), run('r-old', 'orders', 't-daily')]; + const t = toTrigger({ ...daily, status: 'running' }, runs, null); + expect([t.defaultState, t.state, t.type, t.cron, t.timezone]).toEqual(['paused', 'running', 'schedule', '0 6 * * *', 'UTC']); + expect(t.lastRun?.id).toBe('r-new'); + expect(toTrigger(onInventory, runs, null).lastRun).toBeNull(); + }); + + it('reports lateness from evaluation time and hides sensor parts the server lacks', () => { + const tick = toTick({ tick_id: 'k1', trigger_id: 't-daily', epoch: 1, scheduled_for: '2026-11-07T06:00:00Z', evaluated_at: '2026-11-07T06:02:30Z', outcome: 'requested', run_id: 'r-new' }); + expect(tick.lateBySeconds).toBe(150); + const detail = triggerDetailOf(toTrigger(onInventory, [], tick), onInventory, [tick]); + expect(detail.sensor).toMatchObject({ minimumIntervalSeconds: 60, cursorKnown: false }); + expect([detail.declaration, detail.upcoming, detail.evaluations, detail.stateHistory]).toEqual([null, null, null, null]); + expect(triggerDetailOf(toTrigger(daily, [], null), daily, []).sensor).toBeNull(); + }); +}); + +describe('job mapping', () => { + it('links a job to its triggers and newest run, and leaves unreported policy null', () => { + const wire: W['JobRecord'] = { job_key: 'orders', kind: 'job', description: 'Build orders', selection: ['model:customers', 'model:orders'], concurrency_groups: ['warehouse_heavy'], snapshot_id: SNAPSHOT }; + const triggers = [toTrigger(daily, [], null), toTrigger(onInventory, [], null)]; + const job = toJob(wire, triggers, [run('r-inv', 'inventory', null), run('r-ord', 'orders', null)]); + expect(job).toMatchObject({ key: 'orders', kind: 'job', nodeCount: 2, triggerIds: ['t-daily'], concurrencyGroups: ['warehouse_heavy'], select: null, overlap: null, priority: null }); + expect(job.lastRun?.id).toBe('r-ord'); + const detail = jobDetailOf(job, wire); + expect(detail.resolvedNodes).toEqual(['model:customers', 'model:orders']); + expect([detail.settings, detail.concurrency, detail.memberOf, detail.cursorType]).toEqual([null, null, null, null]); + expect(toJob({ ...wire, job_key: 'nightly', kind: 'group' }, triggers, []).kind).toBe('group'); + }); +}); diff --git a/ui/src/lib/api/http-jobs.ts b/ui/src/lib/api/http-jobs.ts new file mode 100644 index 0000000000..500bbd824a --- /dev/null +++ b/ui/src/lib/api/http-jobs.ts @@ -0,0 +1,144 @@ +import type { components } from '$control/schema.d.ts'; +import type { Job, JobDetail, Run, RunRef, Tick, Trigger, TriggerDetail, TriggerKind } from './types/operate'; + +type Wire = components['schemas']; + +const refOf = (run: Run): RunRef => ({ id: run.id, status: run.status, startedAt: run.startedAt, endedAt: run.endedAt }); + +/** The newest listed run matching, as listed newest first. */ +const newest = (runs: Run[], match: (run: Run) => boolean): RunRef | null => { + const run = runs.find(match); + return run ? refOf(run) : null; +}; + +function isSchedule(declaration: Wire['TriggerDeclaration']): declaration is Wire['TriggerDefinition'] { + return 'schedule' in declaration; +} + +export function triggerKindOf(declaration: Wire['TriggerDeclaration']): TriggerKind { + if (isSchedule(declaration)) return declaration.schedule.kind === 'cron' ? 'cron_schedule' : 'calendar_schedule'; + const kind = declaration.when.kind; + if (kind === 'run_event') return 'job_sensor'; + if (kind === 'platform_event') return 'platform_sensor'; + if (kind === 'custom') return 'custom_sensor'; + return 'node_sensor'; +} + +/** The declared cadence or condition in words. */ +export function cadenceOf(declaration: Wire['TriggerDeclaration']): string { + if (isSchedule(declaration)) { + const s = declaration.schedule; + return s.kind === 'cron' ? `${s.expression} (${s.timezone})` : `${s.times.length} calendar time${s.times.length === 1 ? '' : 's'}`; + } + const w = declaration.when; + if (w.kind === 'run_event') return `when ${w.job_key} ${w.kinds.join(' or ')}`; + if (w.kind === 'node_event') return `when ${w.selection.join(', ')} ${w.kinds.join(' or ')}`; + if (w.kind === 'source_new_data') return `when ${w.selection.join(', ')} has new data`; + if (w.kind === 'source_stale') return `when ${w.selection.join(', ')} is stale for ${w.after_seconds}s`; + if (w.kind === 'platform_event') return `on platform events: ${w.kinds.join(', ')}`; + return `custom check ${w.function}`; +} + +export function toTick(wire: Wire['TickRecord']): Tick { + return { + id: wire.tick_id, + scheduledFor: wire.scheduled_for, + evaluatedAt: wire.evaluated_at, + outcome: wire.outcome, + runId: wire.run_id ?? null, + coveredByRunId: wire.covered_by_run_id ?? null, + reason: null, + lateBySeconds: Math.max(0, Math.round((Date.parse(wire.evaluated_at) - Date.parse(wire.scheduled_for)) / 1000)), + catchUp: false + }; +} + +export function toTrigger(wire: Wire['TriggerRecord'], runs: Run[], lastTick: Tick | null): Trigger { + const d = wire.definition_json; + const schedule = isSchedule(d) ? d.schedule : null; + return { + id: wire.trigger_id, + name: wire.name, + type: schedule ? 'schedule' : 'sensor', + kind: triggerKindOf(d), + description: d.description, + jobKey: wire.job_key, + cadence: cadenceOf(d), + cron: schedule?.kind === 'cron' ? schedule.expression : null, + timezone: schedule?.kind === 'cron' ? schedule.timezone : null, + defaultState: d.default_status ?? 'running', + state: wire.status, + nextDueAt: wire.next_due_at ?? null, + lastTick, + lastRun: newest(runs, (r) => r.triggerId === wire.trigger_id), + sourceFile: null, + definitionChanged: false, + warning: null + }; +} + +/** A trigger's detail; parts with no server source are null and their widgets stay hidden. */ +export function triggerDetailOf(trigger: Trigger, wire: Wire['TriggerRecord'], ticks: Tick[]): TriggerDetail { + const d = wire.definition_json; + return { + ...trigger, + declaration: null, + targets: null, + catchUp: null, + upcoming: null, + calendar: null, + sensor: isSchedule(d) + ? null + : { + condition: trigger.cadence, + nodeConditions: [], + minimumIntervalSeconds: d.minimum_interval_seconds, + timeoutSeconds: null, + cursor: null, + cursorUpdatedAt: null, + cursorKnown: false, + currentFingerprint: null, + lastEvaluatedFingerprint: null + }, + ticks, + evaluations: null, + runRequests: null, + stateHistory: null + }; +} + +export function toJob(wire: Wire['JobRecord'], triggers: Trigger[], runs: Run[]): Job { + return { + key: wire.job_key, + kind: wire.kind === 'group' ? 'group' : 'job', + description: wire.description, + // The server stores the resolved node list, not the selector text. + select: null, + members: null, + nodeCount: wire.selection.length, + overlap: null, + onFailure: null, + concurrencyGroups: wire.concurrency_groups, + priority: null, + owner: null, + sourceFile: null, + triggerIds: triggers.filter((t) => t.jobKey === wire.job_key).map((t) => t.id), + lastRun: newest(runs, (r) => r.jobKey === wire.job_key) + }; +} + +export function jobDetailOf(job: Job, wire: Wire['JobRecord']): JobDetail { + return { + ...job, + declaration: null, + exclude: [], + settings: null, + resolvedNodes: wire.selection, + concurrency: null, + targets: null, + groupMembers: null, + criticalNodes: [], + memberOf: null, + cursorType: null + }; +} diff --git a/ui/src/lib/api/http-runs.ts b/ui/src/lib/api/http-runs.ts index 305ffa1848..86fb68311f 100644 --- a/ui/src/lib/api/http-runs.ts +++ b/ui/src/lib/api/http-runs.ts @@ -1,7 +1,7 @@ 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'; +import type { NodeCounts, Run, RunDetail, RunListQuery, RunNode, RunNodeStatus, RunRow, RunRowPage, RunRowQuery, RunTabCounts } from './types/operate'; type Wire = components['schemas']; @@ -75,6 +75,23 @@ export function toRuns(wires: Wire['RunRecord'][], locationId: string): Run[] { return wires.map((w) => toRun(w, locationId, children.get(w.run_id) ?? 0)); } +/** Listed runs under a run-list query, newest first. */ +export function runsMatching(runs: Run[], query: RunListQuery): Run[] { + const matching = runs.filter( + (r) => + (!query.jobKey || r.jobKey === query.jobKey) && + (!query.triggerId || r.triggerId === query.triggerId) && + (!query.parentRunId || r.parentRunId === query.parentRunId) && + (!query.topLevelOnly || !r.parentRunId) && + (!query.statuses?.length || query.statuses.includes(r.status)) && + (!query.kinds?.length || query.kinds.includes(r.kind)) && + (!query.origins?.length || query.origins.includes(r.origin)) && + (!query.since || r.queuedAt >= query.since) && + (!query.until || r.queuedAt < query.until) + ); + return query.limit === undefined ? matching : matching.slice(0, query.limit); +} + 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; diff --git a/ui/src/lib/api/mock/mock-client.ts b/ui/src/lib/api/mock/mock-client.ts index 0e632aba05..49e78f99f0 100644 --- a/ui/src/lib/api/mock/mock-client.ts +++ b/ui/src/lib/api/mock/mock-client.ts @@ -274,6 +274,11 @@ export class MockApiClient implements ApiClient { return this.#operate.launchJob(scope, key, options); } + async launchJobFromScratch(scope: Scope, key: string, options?: JobLaunchOptions): Promise { + this.#location(scope); + return this.#operate.launchJob(scope, key, { ...options, fullRefresh: true }); + } + async listTriggers(scope: Scope): Promise { this.#location(scope); return this.#operate.listTriggers(scope); diff --git a/ui/src/lib/api/mock/operate-store.ts b/ui/src/lib/api/mock/operate-store.ts index 9765fa7a3a..813a2bc267 100644 --- a/ui/src/lib/api/mock/operate-store.ts +++ b/ui/src/lib/api/mock/operate-store.ts @@ -464,7 +464,7 @@ export class MockOperate { return this.#ctx.reply(this.#view(rec, scope)); } - launchJob(scope: Scope, key: string, options: JobLaunchOptions = {}): Promise { + launchJob(scope: Scope, key: string, options: JobLaunchOptions & { fullRefresh?: boolean } = {}): Promise { this.#need('job.run'); const job = jobDef(key); if (!job) throw new ApiError(`Unknown job ${key}`, 404); @@ -1163,6 +1163,7 @@ export class MockOperate { ...def.sensor, cursor: rt.cursor, cursorUpdatedAt: rt.cursorUpdatedAt ? new Date(rt.cursorUpdatedAt).toISOString() : null, + cursorKnown: true, currentFingerprint: currentFp, lastEvaluatedFingerprint: sensorFp } diff --git a/ui/src/lib/api/mock/operate.ts b/ui/src/lib/api/mock/operate.ts index 35237ec9b7..c500962542 100644 --- a/ui/src/lib/api/mock/operate.ts +++ b/ui/src/lib/api/mock/operate.ts @@ -31,7 +31,13 @@ export const SNAPSHOT_GENERATION = 42; // ---- jobs and groups ---- -type JobDef = Omit & { +/** The mock knows every field the server may leave out. */ +type Known = Omit & { [P in K]-?: NonNullable }; + +type JobDef = Known< + Omit, + 'members' | 'overlap' | 'onFailure' | 'priority' | 'sourceFile' | 'declaration' | 'settings' +> & { after: Record; }; diff --git a/ui/src/lib/api/types/operate.ts b/ui/src/lib/api/types/operate.ts index 40bbd8a81a..5d0a831c31 100644 --- a/ui/src/lib/api/types/operate.ts +++ b/ui/src/lib/api/types/operate.ts @@ -483,7 +483,6 @@ export type CursorWindow = { export type JobLaunchOptions = { window?: CursorWindow; - fullRefresh?: boolean; }; /** `selected` reruns exactly the nodes passed with the request. */ @@ -503,15 +502,16 @@ export type Job = { description: string; /** Selector for jobs; null for groups. */ select: string | null; - /** Member jobs for groups. */ - members: string[]; + /** Member jobs for groups; null when the server doesn't say. */ + members: string[] | null; nodeCount: number; - overlap: OverlapPolicy; - onFailure: OnFailure; + /** Null where the server doesn't report it (also `onFailure`, `priority`, `sourceFile`). */ + overlap: OverlapPolicy | null; + onFailure: OnFailure | null; concurrencyGroups: string[]; - priority: number; + priority: number | null; owner: string | null; - sourceFile: string; + sourceFile: string | null; triggerIds: string[]; lastRun: RunRef | null; }; @@ -532,18 +532,19 @@ export type JobGroupMember = { }; export type JobDetail = Job & { - declaration: { sourceFile: string; line: number; code: string }; + /** Null where the server doesn't report it, as for `settings`, `concurrency`, `groupMembers` and `memberOf`. */ + declaration: { sourceFile: string; line: number; code: string } | null; exclude: string[]; /** Settings are CLI flags validated against the command table. */ - settings: Record; + settings: Record | null; resolvedNodes: string[]; - concurrency: ConcurrencyGroup[]; + concurrency: ConcurrencyGroup[] | null; /** Null when the job applies to every target. */ targets: string[] | null; - groupMembers: JobGroupMember[]; + groupMembers: JobGroupMember[] | null; criticalNodes: string[]; /** Groups this job belongs to. */ - memberOf: string[]; + memberOf: string[] | null; /** Nodes with a cursor, so a cursor window applies. */ cursorType: 'timestamp' | 'integer' | null; }; @@ -601,7 +602,7 @@ export type Trigger = { nextDueAt: Iso | null; lastTick: Tick | null; lastRun: RunRef | null; - sourceFile: string; + sourceFile: string | null; /** Code changed since the last evaluation (sensors). */ definitionChanged: boolean; /** The watchdog's warning, e.g. an overdue tick or a low calendar horizon. */ @@ -647,11 +648,12 @@ export type SensorRunRequest = { export type CatchUpPolicy = { mode: 'latest' | 'all'; max: number | null }; export type TriggerDetail = Trigger & { - declaration: { sourceFile: string; line: number; code: string }; + /** Null where the server doesn't report it, as for `upcoming`, `evaluations`, `runRequests` and `stateHistory`. */ + declaration: { sourceFile: string; line: number; code: string } | null; targets: string[] | null; catchUp: CatchUpPolicy | null; /** Upcoming due times (schedules). */ - upcoming: Iso[]; + upcoming: Iso[] | null; /** Calendar schedules: times are computed ahead and stored up to `storedUntil`. */ calendar: { horizon: string; @@ -668,11 +670,13 @@ export type TriggerDetail = Trigger & { timeoutSeconds: number | null; cursor: string | null; cursorUpdatedAt: Iso | null; - currentFingerprint: string; + /** False when the server doesn't report the cursor; resetting it still works. */ + cursorKnown: boolean; + currentFingerprint: string | null; lastEvaluatedFingerprint: string | null; } | null; ticks: Tick[]; - evaluations: SensorEvaluation[]; - runRequests: SensorRunRequest[]; - stateHistory: TriggerStateEvent[]; + evaluations: SensorEvaluation[] | null; + runRequests: SensorRunRequest[] | null; + stateHistory: TriggerStateEvent[] | null; }; diff --git a/ui/src/routes/jobs/+page.svelte b/ui/src/routes/jobs/+page.svelte index 3046ca2745..5f1243506b 100644 --- a/ui/src/routes/jobs/+page.svelte +++ b/ui/src/routes/jobs/+page.svelte @@ -17,8 +17,8 @@ import { isActiveRun, isProblemRun, runStatusLabel, runStatusTone } from '$lib/status'; import { overlapHelp, overlapPlain } from '$lib/operate/labels'; - const jobs = createResource((client, scope) => client.listJobs(scope), ['jobs', 'runs']); - const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers']); + const jobs = createResource((client, scope) => client.listJobs(scope), ['jobs', 'runs', 'snapshot']); + const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers', 'ticks']); const triggerById = $derived(new Map((triggers.data ?? []).map((t) => [t.id, t]))); const canToggle = $derived(app.can('trigger.toggle')); @@ -46,10 +46,14 @@ if (lastRun === 'active' && !(j.lastRun && isActiveRun(j.lastRun.status))) return false; if (lastRun === 'never' && j.lastRun) return false; const q = query.trim().toLowerCase(); - return !q || `${j.key} ${j.description} ${j.select ?? ''} ${j.members.join(' ')}`.toLowerCase().includes(q); + return !q || `${j.key} ${j.description} ${j.select ?? ''} ${(j.members ?? []).join(' ')}`.toLowerCase().includes(q); }) ); const scheduledCount = $derived(all.filter((j) => j.triggerIds.some((id) => triggerById.get(id)?.state === 'running')).length); + // Columns the server doesn't report stay hidden rather than showing dashes. + const showPolicy = $derived(all.some((j) => j.overlap !== null)); + const showOwner = $derived(all.some((j) => j.owner !== null)); + const showSelection = $derived(all.some((j) => j.select !== null || j.members !== null)); const failingCount = $derived(all.filter((j) => j.lastRun && isProblemRun(j.lastRun.status)).length); @@ -96,11 +100,11 @@ Job - Selection + {#if showSelection}Selection{/if} Nodes - If already running + {#if showPolicy}If already running{/if} Started by - Owner + {#if showOwner}Owner{/if} Last run @@ -117,15 +121,17 @@ - - {#if j.kind === 'group'} - Group · {j.members.join(', ')} - {:else} - {j.select} - {/if} - + {#if showSelection} + + {#if j.kind === 'group'} + Group{#if j.members} · {j.members.join(', ')}{/if} + {:else} + {j.select} + {/if} + + {/if} {j.nodeCount} - {overlapPlain[j.overlap]}{#if j.priority} · priority {j.priority}{/if} + {#if showPolicy}{j.overlap ? overlapPlain[j.overlap] : '—'}{#if j.priority} · priority {j.priority}{/if}{/if} {#if jobTriggers.length === 0} Only by hand @@ -151,7 +157,7 @@ {/if} - {j.owner ?? '—'} + {#if showOwner}{j.owner ?? '—'}{/if} {#if j.lastRun} e.stopPropagation()}> diff --git a/ui/src/routes/jobs/[name]/group-order.svelte b/ui/src/routes/jobs/[name]/group-order.svelte index 3ad0c95d05..fe89275eba 100644 --- a/ui/src/routes/jobs/[name]/group-order.svelte +++ b/ui/src/routes/jobs/[name]/group-order.svelte @@ -6,8 +6,8 @@ let { job }: { job: JobDetail } = $props(); const stages = $derived.by(() => { - const max = Math.max(0, ...job.groupMembers.map((m) => m.stage)); - return Array.from({ length: max + 1 }, (_, i) => job.groupMembers.filter((m) => m.stage === i)); + const max = Math.max(0, ...(job.groupMembers ?? []).map((m) => m.stage)); + return Array.from({ length: max + 1 }, (_, i) => (job.groupMembers ?? []).filter((m) => m.stage === i)); }); diff --git a/ui/src/routes/jobs/[name]/job-view.svelte b/ui/src/routes/jobs/[name]/job-view.svelte index 9503871597..e61bec3718 100644 --- a/ui/src/routes/jobs/[name]/job-view.svelte +++ b/ui/src/routes/jobs/[name]/job-view.svelte @@ -14,15 +14,16 @@ let { name }: { name: string } = $props(); - const job = createResource((client, scope) => client.getJob(scope, name), ['jobs', 'runs']); + const job = createResource((client, scope) => client.getJob(scope, name), ['jobs', 'runs', 'snapshot']); const jobRuns = createResource((client, scope) => client.listRuns(scope, { jobKey: name, limit: 12 }), ['runs']); - const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers']); + const triggers = createResource((client, scope) => client.listTriggers(scope), ['triggers', 'ticks']); const nodes = createResource((client, scope) => client.listSnapshotNodes(scope), ['snapshot']); const def = $derived(job.data ?? null); const kindOf = $derived(new Map((nodes.data ?? []).map((n) => [n.key, n.kind]))); const jobTriggers = $derived((triggers.data ?? []).filter((t) => def?.triggerIds.includes(t.id))); const runs = $derived(jobRuns.data ?? []); + const showCost = $derived(runs.some((r) => r.cost !== null)); const canToggle = $derived(app.can('trigger.toggle')); let launchOpen = $state(false); let actionError = $state(null); @@ -68,7 +69,7 @@ ← Jobs {name} {#if def} - {def.kind === 'group' ? `Group of ${def.members.length} jobs` : 'Job'} · {def.nodeCount} nodes + {def.kind === 'group' ? (def.members ? `Group of ${def.members.length} jobs` : 'Group') : 'Job'} · {def.nodeCount} nodes {def.description} {/if} @@ -82,12 +83,14 @@ {#if def}
-
-

Code read-only; change it in the project and deploy

- -
+ {#if def.declaration} +
+

Code read-only; change it in the project and deploy

+ +
+ {/if} - {#if def.kind === 'group'} + {#if def.kind === 'group' && def.groupMembers}

Jobs in this group, in order from the data each job reads and writes, plus any after in the code

@@ -107,8 +110,8 @@ {formatNodeCounts(r.nodeCounts)} {r.startedAt ? formatDuration(r.startedAt, r.endedAt, app.now) : '—'} {formatAgo(r.startedAt ?? r.queuedAt, app.now)} - {r.requestedBy.name} - {formatCost(r.cost)} + {r.requestedBy.displayName} + {#if showCost}{formatCost(r.cost)}{/if} {/each} @@ -119,7 +122,7 @@
-

What it builds · {def.resolvedNodes.length} nodes {def.select ? `--select ${def.select}` : 'everything its jobs build'}{def.exclude.length ? ` --exclude ${def.exclude.join(' ')}` : ''}

+

What it builds · {def.resolvedNodes.length} nodes {def.select ? `--select ${def.select}` : def.kind === 'group' ? 'everything its jobs build' : ''}{def.exclude.length ? ` --exclude ${def.exclude.join(' ')}` : ''}

{#each def.resolvedNodes as n (n)} @@ -138,29 +141,37 @@
Kind
{def.kind === 'group' ? 'Group of jobs' : 'Job'}
{#if def.select}
Selection
{def.select}
{/if} -
If already running
{overlapPlain[def.overlap]}
-
If something fails
{onFailurePlain[def.onFailure]}
-
Priority
{def.priority}
-
Targets
{def.targets?.join(', ') ?? 'all targets'}
-
Owner
{def.owner ?? '—'}
-
Defined in
{def.sourceFile}
- {#if def.memberOf.length} + {#if def.overlap}
If already running
{overlapPlain[def.overlap]}
{/if} + {#if def.onFailure}
If something fails
{onFailurePlain[def.onFailure]}
{/if} + {#if def.priority !== null}
Priority
{def.priority}
{/if} + {#if def.declaration}
Targets
{def.targets?.join(', ') ?? 'all targets'}
{/if} + {#if def.owner}
Owner
{def.owner}
{/if} + {#if def.sourceFile}
Defined in
{def.sourceFile}
{/if} + {#if def.memberOf?.length} {/if}
-

{overlapHelp[def.overlap]}. {onFailureHelp[def.onFailure]}.

+ {#if def.overlap && def.onFailure}

{overlapHelp[def.overlap]}. {onFailureHelp[def.onFailure]}.

{/if} -

Extra options

- {#if Object.keys(def.settings).length} - {#each Object.entries(def.settings) as [flag, value] (flag)} -

{flag}{value === true ? '' : ` ${value}`}

- {/each} - {:else} -

None; it runs with the defaults.

+ {#if def.settings} +

Extra options

+ {#if Object.keys(def.settings).length} + {#each Object.entries(def.settings) as [flag, value] (flag)} +

{flag}{value === true ? '' : ` ${value}`}

+ {/each} + {:else} +

None; it runs with the defaults.

+ {/if} {/if}

Shared slots

- {#if def.concurrency.length} + {#if def.concurrency === null} + {#if def.concurrencyGroups.length} + {#each def.concurrencyGroups as g (g)}

Uses {slotLabel(g)} slots

{/each} + {:else} +

None. It only waits for runs that update the same data.

+ {/if} + {:else if def.concurrency.length} {#each def.concurrency as g (g.key)}

{g.holders.length} of {g.limit} {slotLabel(g.key)} slot{g.limit === 1 ? '' : 's'} in use @@ -173,7 +184,7 @@

Schedules and sensors · {jobTriggers.length}

{#if jobTriggers.length === 0} -

Nothing starts it automatically. Run it by hand{def.memberOf.length ? ', or through its groups' : ''}.

+

Nothing starts it automatically. Run it by hand{def.memberOf?.length ? ', or through its groups' : ''}.

{:else} {#each jobTriggers as t (t.id)}
diff --git a/ui/src/routes/jobs/[name]/launch-form.svelte b/ui/src/routes/jobs/[name]/launch-form.svelte index aa2edd2976..8da2536351 100644 --- a/ui/src/routes/jobs/[name]/launch-form.svelte +++ b/ui/src/routes/jobs/[name]/launch-form.svelte @@ -14,6 +14,7 @@ let error = $state(null); const canRewrite = $derived(app.can('run.rewrite_history')); + const canFullRefresh = $derived(app.client.supports('launchJobFromScratch')); const rewrites = $derived(fullRefresh || useWindow); const placeholder = $derived(job.cursorType === 'integer' ? 'e.g. 1200' : 'e.g. 2026-10-01T00:00:00Z'); const windowInvalid = $derived( @@ -31,10 +32,8 @@ busy = true; error = null; try { - const run = await app.client.launchJob(scope, job.key, { - fullRefresh: fullRefresh || undefined, - window: useWindow && job.cursorType ? { type: job.cursorType, start: start || null, end: end || null } : undefined - }); + const options = { window: useWindow && job.cursorType ? { type: job.cursorType, start: start || null, end: end || null } : undefined }; + const run = fullRefresh ? await app.client.launchJobFromScratch(scope, job.key, options) : await app.client.launchJob(scope, job.key, options); onclose(); await goto(`/runs/${run.id}`); } catch (e) { @@ -48,13 +47,16 @@

Run {job.kind === 'group' ? 'group' : 'job'} {job.key} now

- {job.kind === 'group' ? `Starts a group run, with one run per job in it (${job.members.length}).` : `Builds ${job.nodeCount} nodes with the job's options.`} If another run is updating the same data, it waits for it to finish. + {job.kind === 'group' ? `Starts a group run, with one run per job in it${job.members ? ` (${job.members.length})` : ''}.` : `Builds ${job.nodeCount} nodes with the job's options.`} If another run is updating the same data, it waits for it to finish.

+ {#if canFullRefresh || job.cursorType}
+ {#if canFullRefresh} + {/if} {#if job.cursorType}
+ {/if} {#if error}

{error}

{/if}
diff --git a/ui/src/routes/triggers/[id]/sensor-panel.svelte b/ui/src/routes/triggers/[id]/sensor-panel.svelte index b85ec51d57..8064bca98e 100644 --- a/ui/src/routes/triggers/[id]/sensor-panel.svelte +++ b/ui/src/routes/triggers/[id]/sensor-panel.svelte @@ -54,9 +54,13 @@

Where it's up to

- {sensor.cursor ?? '(the start: nothing handled yet)'} - {#if sensor.cursorUpdatedAt}updated {formatAgo(sensor.cursorUpdatedAt, app.now)}{/if} - (confirmReset = !confirmReset)}>Start over + {#if sensor.cursorKnown} + {sensor.cursor ?? '(the start: nothing handled yet)'} + {#if sensor.cursorUpdatedAt}updated {formatAgo(sensor.cursorUpdatedAt, app.now)}{/if} + {:else} + Starting over makes the next check look at everything again. + {/if} + (confirmReset = !confirmReset)}>Start over
{#if confirmReset}
@@ -71,6 +75,7 @@ {#if error}

{error}

{/if}
+{#if trigger.evaluations}

Checks · {trigger.evaluations.length}

@@ -95,7 +100,9 @@
{#if trigger.evaluations.length > shown}{/if}
+{/if} +{#if trigger.runRequests}

Runs it started · {trigger.runRequests.length}

{#if trigger.runRequests.length} @@ -117,6 +124,7 @@

{trigger.jobKey ? 'It hasn’t started any runs yet.' : 'This sensor only runs its own code; it doesn’t start runs.'}

{/if}
+{/if}