feat: column lineage, worker ad-hoc commands and UI pages

.github/workflows/verify.yml

@@ -174,9 +174,10 @@ jobs:
                 ;;
             esac
             case "${path}" in
-              ui/*|clients/typescript/*|.github/workflows/verify.yml)
+              ui/*|clients/typescript/*|.github/workflows/verify.yml|website/src/assets/logo-light.png|website/src/assets/logo-dark.png|website/public/favicon.png)
                 # The web UI's own checks and unit tests; the wheels embed its build. It compiles
-                # against the generated control client, so client changes run them too.
+                # against the generated control client, so client changes run them too, and it
+                # imports the website's logos and favicon, so moving them fails its build.
                 ui=true
                 ;;
             esac
@@ -357,11 +358,6 @@ jobs:
           node-version: "22"
           cache: npm
           cache-dependency-path: clients/typescript/package-lock.json
-      - name: Check generated control client against the built native API
-        run: |
-          npm --prefix clients/typescript ci
-          uv run --no-sync npm --prefix clients/typescript run check
-          uv run --no-sync npm --prefix clients/typescript test
       - name: Fetch the base for the new-duplicate check
         run: git fetch --no-tags --depth=1 origin "$BASE_REF"
         env:

Makefile

@@ -1,6 +1,6 @@
 SHELL := /bin/bash
 
-.PHONY: verify verify-quick verify-pg coverage cli-preview rust-check check-native-layering check-dupes-new check-native-stub check-build-features check-python-boundary check-release-gate check-docs-skill \
+.PHONY: verify verify-quick verify-pg coverage cli-preview rust-check check-native-layering check-dupes-new check-native-stub check-build-features check-python-boundary check-release-gate check-docs-skill check-control-client \
 	check-e2e-shards test-e2e-duckdb test-e2e-duckdb-build-core \
 	test-e2e-duckdb-build-incremental \
 	test-e2e-duckdb-cli-data test-e2e-duckdb-cli test-e2e-duckdb-inspection \
@@ -706,14 +706,21 @@ verify-quick:
 	exit $$status
 
 
-check-ci: check-dupes-new check-e2e-shards check-native-layering check-native-stub check-build-features check-python-boundary check-release-gate check-docs-skill rust-lint
+check-ci: check-dupes-new check-e2e-shards check-native-layering check-native-stub check-build-features check-python-boundary check-release-gate check-docs-skill check-control-client rust-lint
 	uv run ruff format --check .
 	uv run ruff check .
 	uv run ty check src tests
 	uv run pytest tests/unit/src/sqlbuild/adapter/contract/classes/strict_adapter/test_strict_adapter.py -q
 	uv run fensu check
 
 
+# The generated TypeScript control client must match the OpenAPI document of the installed sqb.
+check-control-client:
+	npm --prefix clients/typescript ci
+	uv run --no-sync npm --prefix clients/typescript run check
+	uv run --no-sync npm --prefix clients/typescript test
+
+
 check-native-layering:
 	uv run python -m scripts.check_native_layering
 

clients/typescript/src/schema.d.ts

@@ -338,6 +338,22 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/api/control/locations/{location_id}/targets/{target}/column-lineage": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get: operations["readPersonColumnLineage"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/api/control/locations/{location_id}/targets/{target}/connection-tests": {
         parameters: {
             query?: never;
@@ -1334,6 +1350,21 @@ export interface components {
         CliLoginStartRequest: {
             client_hash: string;
         };
+        /** @description One published column lineage edge; `transform` is the compiler's transform kind, such as `direct` or `cast`. */
+        ColumnLineageEdge: {
+            source_column: string;
+            source_key: string;
+            target_column: string;
+            target_key: string;
+            transform: string;
+        };
+        /** @description The column lineage reaching or leaving one node, from the current snapshot; empty before one publishes lineage. */
+        ColumnLineageRead: {
+            edges: components["schemas"]["ColumnLineageEdge"][];
+            nodes: components["schemas"]["LineageNodeColumns"][];
+            /** Format: uuid */
+            snapshot_id?: string | null;
+        };
         /** @description A row whose key was already stored with different content; the first row is kept. */
         ConflictEvidence: {
             existing_sha256: string;
@@ -1472,6 +1503,16 @@ export interface components {
             target: string;
             window?: null | components["schemas"]["RunWindow"];
         };
+        LineageColumn: {
+            data_type?: string | null;
+            name: string;
+        };
+        /** @description A node's columns as its compiled definition infers them; `data_type` is absent when inference could not tell. */
+        LineageNodeColumns: {
+            columns: components["schemas"]["LineageColumn"][];
+            kind: string;
+            node_key: string;
+        };
         /** @description One location target the person can view, with the actions allowed on at least one of its nodes. */
         LocationAccess: {
             actions: components["schemas"]["Action"][];
@@ -1989,6 +2030,11 @@ export interface components {
             claim_id: string;
             request: components["schemas"]["RunRequest"];
         };
+        /**
+         * @description The `sqb` command a run executes; job and group runs always build.
+         * @enum {string}
+         */
+        RunCommand: "build" | "test" | "seed" | "audit" | "freshness";
         RunErrorText: {
             error_type?: string | null;
             invocation_id: string;
@@ -2121,6 +2167,7 @@ export interface components {
         };
         /** @description The authorized selection and pinned execution inputs are immutable once requested. */
         RunRequest: {
+            command?: components["schemas"]["RunCommand"];
             concurrency_groups: string[];
             image_id?: string | null;
             job_key?: string | null;
@@ -3379,6 +3426,46 @@ export interface operations {
             };
         };
     };
+    readPersonColumnLineage: {
+        parameters: {
+            query: {
+                /** @description The node whose upstream and downstream column lineage to return */
+                node_key: string;
+            };
+            header?: never;
+            path: {
+                location_id: string;
+                target: string;
+            };
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ColumnLineageRead"];
+                };
+            };
+            /** @description Session rejected */
+            401: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content?: never;
+            };
+            403: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["ApiErrorBody"];
+                };
+            };
+        };
+    };
     requestConnectionTest: {
         parameters: {
             query?: never;

crates/sqlbuild-cli/src/adhoc/_helpers/planner.rs

@@ -11,9 +11,10 @@ use sqlbuild_orchestration::runs::models::{CommandPlan, PlannerResource};
 use sqlbuild_orchestration::runs::types::CommandPlanner;
 
 use crate::adhoc::constants::{
-    ADHOC_VERBS, ALLOWED_DESTS, BUILD_VERB, COMMAND_DEST, INTEGER_CONVERTERS, LOAD_SOURCES_DEST,
-    POSITIVE_INTEGER_CONVERTER, REFUSED_DESTS, SELECTOR_DESTS,
+    ADHOC_VERBS, ALLOWED_DESTS, BUILD_VERB, COMMAND_DEST, FRESHNESS_VERB, INTEGER_CONVERTERS,
+    LOAD_SOURCES_DEST, POSITIVE_INTEGER_CONVERTER, REFUSED_DESTS, SEED_VERB, SELECTOR_DESTS,
 };
+use crate::argv::main::command_settings_arguments::command_settings_arguments;
 use crate::argv::main::job_settings_arguments::job_settings_arguments;
 use crate::argv::main::parse_cli_arguments::parse_cli_arguments;
 use crate::argv::models::{ArgvSettings, Namespace, TypeFailure, Value};
@@ -52,13 +53,29 @@ impl CommandPlanner for CliPlanner {
         }
         if verb == BUILD_VERB {
             job_settings_arguments(&settings)?;
+        } else {
+            command_settings_arguments(verb, &settings)?;
         }
         let mut select: Vec<String> = selectors.remove("select").unwrap_or_default();
         select.extend(selectors.remove("positional_select").unwrap_or_default());
         let exclude: Vec<String> = selectors.remove("exclude").unwrap_or_default();
+        let selection: BTreeSet<String> = match verb.as_str() {
+            SEED_VERB => {
+                let seeds: BTreeSet<String> = resolve(resources, &select, &exclude)?
+                    .into_iter()
+                    .filter(|key| key.starts_with("seed:"))
+                    .collect();
+                if seeds.is_empty() {
+                    return Err("the selectors match no seeds in the current publication".into());
+                }
+                seeds
+            }
+            FRESHNESS_VERB => freshness_sources(resources, &select, &exclude)?,
+            _ => resolve(resources, &select, &exclude)?,
+        };
         Ok(CommandPlan {
             verb: verb.clone(),
-            selection: resolve(resources, &select, &exclude)?,
+            selection,
             settings,
         })
     }
@@ -163,18 +180,69 @@ fn resolve(
     resources: &[PlannerResource],
     select: &[String],
     exclude: &[String],
+) -> Result<BTreeSet<String>, String> {
+    let selected: BTreeSet<String> = resolved(resources, select, exclude)?;
+    if selected.is_empty() {
+        return Err("the selectors match no nodes in the current publication".into());
+    }
+    Ok(selected)
+}
+
+fn resolved(
+    resources: &[PlannerResource],
+    select: &[String],
+    exclude: &[String],
 ) -> Result<BTreeSet<String>, String> {
     let graph_resources: Vec<GraphResource> = resources.iter().filter_map(graph_resource).collect();
     let graph = build_project_graph(&graph_resources);
-    let selected: BTreeSet<String> = resolve_selectors(&graph, select, exclude)
+    Ok(resolve_selectors(&graph, select, exclude)
         .map_err(|error| format!("{}: {}", error.code, error.message))?
         .into_iter()
         .map(|(kind, name)| format!("{kind}:{name}"))
+        .collect())
+}
+
+/// Sources `sqb freshness` observes: selected or upstream sources, less those the exclusions reach.
+fn freshness_sources(
+    resources: &[PlannerResource],
+    select: &[String],
+    exclude: &[String],
+) -> Result<BTreeSet<String>, String> {
+    let selected: BTreeSet<String> =
+        upstream_sources(resources, &resolved(resources, select, &[])?);
+    let excluded: BTreeSet<String> = if exclude.is_empty() {
+        BTreeSet::new()
+    } else {
+        upstream_sources(resources, &resolved(resources, exclude, &[])?)
+    };
+    let sources: BTreeSet<String> = selected.difference(&excluded).cloned().collect();
+    if sources.is_empty() {
+        return Err("no sources are selected or upstream of the selection".into());
+    }
+    Ok(sources)
+}
+
+fn upstream_sources(resources: &[PlannerResource], keys: &BTreeSet<String>) -> BTreeSet<String> {
+    let deps: BTreeMap<&str, &[String]> = resources
+        .iter()
+        .map(|resource| (resource.key.as_str(), resource.deps.as_slice()))
         .collect();
-    if selected.is_empty() {
-        return Err("the selectors match no nodes in the current publication".into());
+    let mut reached: BTreeSet<&str> = BTreeSet::new();
+    let mut pending: Vec<&str> = keys.iter().map(String::as_str).collect();
+    while let Some(key) = pending.pop() {
+        if reached.insert(key) {
+            pending.extend(
+                deps.get(key)
+                    .into_iter()
+                    .flat_map(|deps| deps.iter().map(String::as_str)),
+            );
+        }
     }
-    Ok(selected)
+    reached
+        .into_iter()
+        .filter(|key| key.starts_with("source:"))
+        .map(str::to_owned)
+        .collect()
 }
 
 /// A published node as the selector graph sees it; keys without a kind are not selectable.

crates/sqlbuild-cli/src/adhoc/constants.rs

@@ -6,6 +6,12 @@ pub(crate) const ADHOC_VERBS: [&str; 5] = ["build", "test", "seed", "audit", "fr
 /// The verb whose settings must also form valid job settings.
 pub(crate) const BUILD_VERB: &str = "build";
 
+/// The verb that runs only the seeds of its selection.
+pub(crate) const SEED_VERB: &str = "seed";
+
+/// The verb that observes sources, including those upstream of selected models.
+pub(crate) const FRESHNESS_VERB: &str = "freshness";
+
 /// The parser destination holding the verb itself, which is not a setting.
 pub(crate) const COMMAND_DEST: &str = "command";
 

crates/sqlbuild-cli/src/adhoc/tests/test_adhoc_planner.rs

@@ -180,6 +180,34 @@ fn given_ad_hoc_commands_when_planned_then_flags_follow_the_allowlist_and_select
             command: &["test", "-s", "orders", "--case", "happy_path"],
             expected_plan: Ok((&["model:orders"], json!({"case": "happy_path"}))),
         },
+        AdhocPlanTestCase {
+            description: "audit settings come from the audit parser",
+            command: &["audit", "-s", "orders", "--concurrency", "2"],
+            expected_plan: Ok((&["model:orders"], json!({"concurrency": 2}))),
+        },
+        AdhocPlanTestCase {
+            description: "seed runs only the seeds its selectors reach",
+            command: &["seed", "+orders"],
+            expected_plan: Ok((&["seed:currencies"], json!({}))),
+        },
+        AdhocPlanTestCase {
+            description: "seed refuses a selection without seeds, which would run nothing",
+            command: &["seed", "orders"],
+            expected_plan: Err("the selectors match no seeds"),
+        },
+        AdhocPlanTestCase {
+            description: "freshness observes every source upstream of a selected model",
+            command: &["freshness", "order_totals", "--state", "--fail-on-stale"],
+            expected_plan: Ok((
+                &["source:shop.raw_orders"],
+                json!({"fail_on_stale": true, "state": true}),
+            )),
+        },
+        AdhocPlanTestCase {
+            description: "freshness excludes the sources an excluded model reads",
+            command: &["freshness", "--exclude", "orders"],
+            expected_plan: Err("no sources are selected or upstream of the selection"),
+        },
     ];
     let planner = adhoc_planner();
     for test_case in test_cases {

crates/sqlbuild-cli/src/argv/main/command_settings_arguments.rs

@@ -0,0 +1,30 @@
+//! Turn frozen settings back into one command's flags, against the same table as the command-line parser.
+
+use std::collections::BTreeMap;
+
+use serde_json::Value as Json;
+
+use crate::argv::_helpers::job_settings::{setting_argument, validate_mutex};
+use crate::argv::_helpers::table::root_parser;
+
+/// Produce argument tokens for `command`, never shell text; a setting the command lacks is refused.
+pub fn command_settings_arguments(
+    command: &str,
+    settings: &BTreeMap<String, Json>,
+) -> Result<Vec<String>, String> {
+    let parser = root_parser()
+        .actions
+        .iter()
+        .flat_map(|action| &action.parsers)
+        .find(|(name, _)| name == command)
+        .map(|(_, parser)| parser)
+        .ok_or_else(|| format!("{command} command is unavailable"))?;
+    let mut arguments: Vec<String> = Vec::new();
+    for (name, value) in settings {
+        if let Some(argument) = setting_argument(parser, name, value)? {
+            arguments.push(argument);
+        }
+    }
+    validate_mutex(parser, &arguments)?;
+    Ok(arguments)
+}