@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "crates/sqlbuild-core",
+    "crates/sqlbuild-cursor",
     "crates/sqlbuild-pysource",
     "crates/sqlbuild-cache",
     "crates/sqlbuild-sqltext",
@@ -24,6 +25,7 @@ resolver = "2"
 [workspace.metadata.native-layers]
 order = [
     "sqlbuild-core",
+    "sqlbuild-cursor",
     "sqlbuild-pysource",
     "sqlbuild-cache",
     "sqlbuild-sqltext",
@@ -93,6 +95,7 @@ sqlbuild-cache = { path = "crates/sqlbuild-cache" }
 sqlbuild-cli = { path = "crates/sqlbuild-cli" }
 sqlbuild-config = { path = "crates/sqlbuild-config" }
 sqlbuild-core = { path = "crates/sqlbuild-core" }
+sqlbuild-cursor = { path = "crates/sqlbuild-cursor" }
 sqlbuild-discovery = { path = "crates/sqlbuild-discovery" }
 sqlbuild-render = { path = "crates/sqlbuild-render" }
 sqlbuild-refactor = { path = "crates/sqlbuild-refactor" }
 
@@ -7,10 +7,11 @@ that can be tested and benchmarked with `cargo test` alone.
 | Crate | Responsibility |
 |---|---|
 | `sqlbuild-core` | Constants shared by several crates, the panic boundary that turns an unwinding native failure into a named compiler error, passive identity, diagnostic and SQL value types, text positions, and a JSON emitter whose output matches Python's `json.dumps` and `orjson` byte for byte. |
+| `sqlbuild-cursor` | Microbatch cursor algebra: typed cursor values, grain flooring and stepping, half-open interval clamping, merging and batch splitting, timestamp text normalization and calendar durations, with Python `datetime` results and errors. Zones other than fixed offsets are resolved through a caller-supplied lookup. |
 | `sqlbuild-pysource` | Static reading of project Python source with the ruff parser: macro module structure, literal SQL arguments, definition locations and string-literal edits. It never imports or runs user code; CPython stays the authority on whether a file is valid Python. |
 | `sqlbuild-cache` | The shared native store every native cache uses (one atomically replaced, checksummed file per cache kind, keyed by content digests) and the one fingerprinting scheme for cache keys: the content digest and the project file fingerprint. |
 | `sqlbuild-sqltext` | Lexical SQL text without polyglot: comment, quote and parenthesis scanning, the rules quote policy, model header tokenization and matching, static variable substitution, static reference extraction, and the general reference scan with table-function call arguments under each adapter's lexical rules. |
-| `sqlbuild-config` | Configuration file reading without Python: TOML, YAML 1.2 with the core schema and merge keys, and typed project and local config readers for the fields discovery needs. Errors carry their kind and position; YAML outside the supported subset is an `Unsupported` error. |
+| `sqlbuild-config` | Configuration file reading without Python: TOML, YAML 1.2 with the core schema and merge keys, and typed project and local config readers for the fields discovery needs. It also writes the workspace files `sqb playground` and `sqb skills` create from packaged templates and skills. Errors carry their kind and position; YAML outside the supported subset is an `Unsupported` error. |
 | `sqlbuild-discovery` | Native project discovery and the only implementation of the model, SQL test, scenario, source and schema file collections and the declaration layout: the shared directory walk with Python's glob and sort semantics, file reading with Python's newline handling and read errors, and parsing of authored files with Python's exact discovery messages. Results are plain data the Python discovery facade materialises. |
 | `sqlbuild-scopes` | Declaration scopes, visibility, relationship grants, placement, report queries and the offline scope cache, including dialect-aware SQL-test and scenario scans. |
 | `sqlbuild-model-config` | MODEL configuration, columns and audits, template expansion over caller values, and dynamic-column name patterns using Rust regex syntax. |
@@ -30,7 +31,7 @@ Dependencies point one way, from the top of this graph to the bottom:
 sqlbuild-python
   -> sqlbuild-refactor -> sqlbuild-rules -> sqlbuild-analysis -> sqlbuild-attachments
   -> sqlbuild-model-config -> sqlbuild-scopes -> sqlbuild-render -> sqlbuild-discovery -> sqlbuild-config
-  -> sqlbuild-sqltext -> sqlbuild-cache -> sqlbuild-pysource -> sqlbuild-core
+  -> sqlbuild-sqltext -> sqlbuild-cache -> sqlbuild-pysource -> sqlbuild-cursor -> sqlbuild-core
 sqlbuild-cli -> sqlbuild-cache, sqlbuild-core
 ```
 
 
@@ -16,6 +16,7 @@ pub mod explicit_references;
 pub mod fingerprint_storage;
 pub mod graph;
 pub mod lineage;
+pub mod lineage_command;
 pub mod lineage_graph;
 pub mod manifest;
 pub mod python_plan;
 
@@ -0,0 +1,175 @@
+//! Human column lineage: dependency lists and column-trace trees.
+
+use std::collections::{HashMap, HashSet};
+use std::convert::Infallible;
+
+use sqlbuild_core::cli_text::models::CliStyle;
+
+use crate::lineage_command::constants::{
+    ALREADY_SHOWN, DOWNSTREAM_DIRECTION, HUMAN_COLUMN_TRACE_LIMIT, STAR_TRANSFORM,
+    STAR_TRANSFORM_LABEL,
+};
+use crate::lineage_command::models::{ColumnTrace, LineageColumn, LineageColumnEdge};
+use crate::lineage_graph::main::trace_limit_note::trace_limit_note;
+use crate::lineage_graph::main::tree_lines::tree_lines;
+use crate::lineage_graph::models::TreeLine;
+use crate::lineage_graph::types::TreeVisitor;
+
+/// `resource.column`.
+fn column_id(column: &LineageColumn) -> String {
+    format!("{}.{}", column.resource_name, column.column_name)
+}
+
+fn transform_label(edge: &LineageColumnEdge) -> &str {
+    if edge.transform == STAR_TRANSFORM {
+        STAR_TRANSFORM_LABEL
+    } else {
+        &edge.transform
+    }
+}
+
+fn displayed(trace: &ColumnTrace) -> &[LineageColumnEdge] {
+    &trace.trace[..trace.trace.len().min(HUMAN_COLUMN_TRACE_LIMIT)]
+}
+
+fn limit_note(trace: &ColumnTrace, style: CliStyle) -> Vec<String> {
+    let notes: Vec<String> = trace_limit_note(
+        i64::try_from(trace.trace.len()).unwrap_or(i64::MAX),
+        i64::try_from(HUMAN_COLUMN_TRACE_LIMIT).unwrap_or(i64::MAX),
+    );
+    if notes.is_empty() {
+        return notes;
+    }
+    let mut lines: Vec<String> = vec![String::new()];
+    lines.extend(notes.iter().map(|note| style.muted(note)));
+    lines
+}
+
+/// The trace as `source -> target  transform` lines.
+pub(crate) fn column_list(trace: &ColumnTrace, style: CliStyle) -> String {
+    if trace.trace.is_empty() {
+        return format!(
+            "Column dependencies\n\n{}",
+            style.section(&column_id(&trace.target))
+        );
+    }
+    let edges: &[LineageColumnEdge] = displayed(trace);
+    let width = |side: fn(&LineageColumnEdge) -> &LineageColumn| -> usize {
+        edges
+            .iter()
+            .map(|edge| column_id(side(edge)).chars().count())
+            .max()
+            .unwrap_or(0)
+    };
+    let left: usize = width(|edge| &edge.source);
+    let right: usize = width(|edge| &edge.target);
+    let mut lines: Vec<String> = vec![style.title("Column dependencies"), String::new()];
+    for edge in edges {
+        let source: String = column_id(&edge.source);
+        let target: String = column_id(&edge.target);
+        lines.push(format!(
+            "{}{} {} {}{} {}",
+            style.section(&source),
+            " ".repeat(left - source.chars().count()),
+            style.muted("->"),
+            style.section(&target),
+            " ".repeat(right - target.chars().count()),
+            style.muted(transform_label(edge))
+        ));
+    }
+    lines.extend(limit_note(trace, style));
+    lines.join("\n")
+}
+
+/// The trace as a tree from the target column.
+pub(crate) fn column_tree(trace: &ColumnTrace, style: CliStyle) -> String {
+    let target: String = column_id(&trace.target);
+    let mut lines: Vec<String> = vec![
+        format!(
+            "{}  {}  {}",
+            style.title("Column trace"),
+            style.section(&target),
+            style.muted(&trace.direction)
+        ),
+        String::new(),
+    ];
+    if trace.trace.is_empty() {
+        lines.push(style.muted("  No column dependencies found"));
+        return lines.join("\n");
+    }
+    let downstream: bool = trace.direction == DOWNSTREAM_DIRECTION;
+    let mut deps: HashMap<String, Vec<&LineageColumnEdge>> = HashMap::new();
+    for edge in displayed(trace) {
+        let key: &LineageColumn = if downstream {
+            &edge.source
+        } else {
+            &edge.target
+        };
+        deps.entry(column_id(key)).or_default().push(edge);
+    }
+    let mut tree = ColumnTree {
+        deps: &deps,
+        downstream,
+        style,
+        lines: Vec::new(),
+    };
+    let roots: Vec<&LineageColumnEdge> = tree.sorted_edges(&target);
+    let Ok(()) = tree_lines(roots, HashSet::from([target]), &mut tree);
+    lines.extend(tree.lines);
+    lines.extend(limit_note(trace, style));
+    lines.join("\n")
+}
+
+/// Column-trace edges shown once per related column, sorted by that column.
+struct ColumnTree<'a> {
+    deps: &'a HashMap<String, Vec<&'a LineageColumnEdge>>,
+    downstream: bool,
+    style: CliStyle,
+    lines: Vec<String>,
+}
+
+impl<'a> ColumnTree<'a> {
+    fn related_id(&self, edge: &LineageColumnEdge) -> String {
+        column_id(if self.downstream {
+            &edge.target
+        } else {
+            &edge.source
+        })
+    }
+
+    fn sorted_edges(&self, id: &str) -> Vec<&'a LineageColumnEdge> {
+        let mut edges: Vec<&'a LineageColumnEdge> = self.deps.get(id).cloned().unwrap_or_default();
+        edges.sort_by_cached_key(|edge| self.related_id(edge));
+        edges
+    }
+}
+
+impl<'a> TreeVisitor<&'a LineageColumnEdge, String, Infallible> for ColumnTree<'a> {
+    fn key(&mut self, edge: &&'a LineageColumnEdge) -> Result<String, Infallible> {
+        Ok(self.related_id(edge))
+    }
+
+    fn line(&mut self, line: TreeLine<'_, &'a LineageColumnEdge>) -> Result<(), Infallible> {
+        let suffix: String = if line.repeated {
+            self.style.muted(ALREADY_SHOWN)
+        } else {
+            String::new()
+        };
+        let edge: &LineageColumnEdge = line.node;
+        self.lines.push(format!(
+            "{}{}{} {}{suffix}",
+            line.prefix,
+            self.style.muted(line.branch),
+            self.style.section(&self.related_id(edge)),
+            self.style.muted(&format!("({})", transform_label(edge)))
+        ));
+        Ok(())
+    }
+
+    fn children(
+        &mut self,
+        edge: &&'a LineageColumnEdge,
+    ) -> Result<Vec<&'a LineageColumnEdge>, Infallible> {
+        Ok(self.sorted_edges(&self.related_id(edge)))
+    }
+}
@@ -0,0 +1,102 @@
+//! Breadth-first closures and selected-graph assembly over a lineage index.
+
+use std::collections::{HashMap, HashSet, VecDeque};
+
+use crate::lineage_command::constants::{
+    BOTH_DIRECTIONS, DOWNSTREAM_DIRECTION, UPSTREAM_DIRECTION,
+};
+use crate::lineage_command::models::{LineageGraph, LineageIndex, LineageKey, LineageNode};
+
+/// Edge lists by key.
+pub(crate) type EdgeMap<'a> = HashMap<&'a LineageKey, &'a [LineageKey]>;
+
+/// `edges` as a lookup by key.
+pub(crate) fn edge_map(edges: &[(LineageKey, Vec<LineageKey>)]) -> EdgeMap<'_> {
+    edges
+        .iter()
+        .map(|(key, targets)| (key, targets.as_slice()))
+        .collect()
+}
+
+/// Keys within `max_depth` edges of any start; a start is included only when reached again.
+pub(crate) fn closure(
+    edges: &EdgeMap<'_>,
+    starts: &[LineageKey],
+    max_depth: Option<i64>,
+) -> HashSet<LineageKey> {
+    let mut visited: HashSet<LineageKey> = HashSet::new();
+    let mut frontier: VecDeque<(&LineageKey, i64)> =
+        starts.iter().map(|start| (start, 0)).collect();
+    while let Some((current, depth)) = frontier.pop_front() {
+        if max_depth.is_some_and(|maximum| depth >= maximum) {
+            continue;
+        }
+        for neighbor in edges.get(current).copied().unwrap_or_default() {
+            if visited.insert(neighbor.clone()) {
+                frontier.push_back((neighbor, depth + 1));
+            }
+        }
+    }
+    visited
+}
+
+/// The keys `direction` reaches from `keys` within `depth`.
+pub(crate) fn expand(
+    index: &LineageIndex,
+    keys: &[LineageKey],
+    (direction, depth): (&str, Option<i64>),
+) -> HashSet<LineageKey> {
+    let mut expanded: HashSet<LineageKey> = HashSet::new();
+    if direction == UPSTREAM_DIRECTION || direction == BOTH_DIRECTIONS {
+        expanded.extend(closure(&edge_map(&index.upstream), keys, depth));
+    }
+    if direction == DOWNSTREAM_DIRECTION || direction == BOTH_DIRECTIONS {
+        expanded.extend(closure(&edge_map(&index.downstream), keys, depth));
+    }
+    expanded
+}
+
+/// `keys` sorted by resource type, then name.
+pub(crate) fn sorted_keys<'a>(keys: impl IntoIterator<Item = &'a LineageKey>) -> Vec<LineageKey> {
+    let mut sorted: Vec<LineageKey> = keys.into_iter().cloned().collect();
+    sorted.sort();
+    sorted.dedup();
+    sorted
+}
+
+/// Display nodes and the upstream edges between selected keys.
+pub(crate) fn build_graph(
+    index: &LineageIndex,
+    selected: &HashSet<LineageKey>,
+    (focus_keys, direction): (Vec<LineageKey>, Option<&str>),
+) -> LineageGraph {
+    let ordered: Vec<LineageKey> = sorted_keys(selected);
+    let upstream: EdgeMap<'_> = edge_map(&index.upstream);
+    let mut edges: Vec<(LineageKey, LineageKey)> = Vec::new();
+    for downstream in &ordered {
+        for upstream_key in upstream.get(downstream).copied().unwrap_or_default() {
+            if selected.contains(upstream_key) {
+                edges.push((upstream_key.clone(), downstream.clone()));
+            }
+        }
+    }
+    LineageGraph {
+        nodes: ordered
+            .iter()
+            .map(|key| {
+                index
+                    .nodes
+                    .get(key)
+                    .cloned()
+                    .unwrap_or_else(|| LineageNode {
+                        key: key.clone(),
+                        relative_path: None,
+                        qualified_name: None,
+                    })
+            })
+            .collect(),
+        edges,
+        focus_keys,
+        direction: direction.map(str::to_owned),
+    }
+}