diff --git a/SPEC.md b/SPEC.md index 350fb43..1195758 100644 --- a/SPEC.md +++ b/SPEC.md @@ -1271,6 +1271,20 @@ a bare JSON array of spec-valid `JobEvent`s and `DatasetEvent`s, each with `eventTime`/`producer`/`schemaURL`, usable directly as a batch-endpoint payload. Both accept a model target or selector terms for scoping. +`lineage --diff ` is the semantic complement to the data-side +`branch_diff`: the workspace subtree at `merge-base(ref, HEAD)` is +materialised read-only (`ls-tree`/`cat-file`, no worktree or checkout +mutation), compiled with the same options, and the two graphs are compared +— nodes and edges added, removed or field-changed, plus the consumers each +removed node orphans and the downstream each removed or moved edge affects. +Edges compare as multisets: parallel edges between the same nodes keep +their metadata distinct. `lineage --diff ` compares two +exact refs instead. The report persists to `lineage_diff.json` with the +resolved base commit, the candidate's git head and worktree state, and the +Nessie branch-pair binding when `--ref`/`--from` resolves — `promote` +audits that provenance and reports the delta as `current`, `advisory`, or +`stale` with the reason rather than treating a moved report as evidence. + ## 56. Column lineage ```bash diff --git a/crates/phlo-transform-cli/src/main.rs b/crates/phlo-transform-cli/src/main.rs index 82b6bd4..f053961 100644 --- a/crates/phlo-transform-cli/src/main.rs +++ b/crates/phlo-transform-cli/src/main.rs @@ -24,11 +24,12 @@ use phlo_transform_engine::{ adapter_default_schema, branch_diff, changed_models, collect_source_states, diff, diff_reasons, ensure_environment, evaluate_gates, materialized_for_environment, model_keys, promote, relation_for_source, retarget, seeds_for_environment, Adapter, ArtifactWriter, - BranchDiffReport, BranchDiffRequest, CancelHandle, ContractSafety, DatasetKind, DiffPolicy, - DiffRequest, DiffStrategy, EnvironmentArtifact, EnvironmentSetup, EnvironmentSpec, - ExecutionStatus, GateInput, Membership, Plan, PlanAction, PlanOptions, PlanReason, Planner, - PostgresStateStore, PromotionRequest, ReasonKind, RetryPolicy, RunOptions, RunResult, Runner, - SqliteStateStore, StateStore, SCHEMA_VERSION, + BranchDiffReport, BranchDiffRequest, CancelHandle, CandidateProvenance, ContractSafety, + DatasetKind, DiffPolicy, DiffRequest, DiffStrategy, EnvironmentArtifact, EnvironmentSetup, + EnvironmentSpec, ExecutionStatus, GateInput, LineageDiffArtifact, LineageEnvironment, + Membership, Plan, PlanAction, PlanOptions, PlanReason, Planner, PostgresStateStore, + PromotionRequest, ReasonKind, RetryPolicy, RunOptions, RunResult, Runner, SqliteStateStore, + StateStore, SCHEMA_VERSION, }; use phlo_transform_nessie::{NessieClient, NessieConfig, NessieRestClient}; use phlo_transform_trino::{TrinoAdapter, TrinoConfig}; @@ -283,6 +284,15 @@ enum Command { /// JSON array of OpenLineage JobEvents and DatasetEvents. #[arg(long, value_enum)] format: Option, + /// Diff lineage against the lineage compiled at a Git baseline. + /// One ref — `lineage --diff main` — uses the same merge-base + /// semantics as `--since`: the workspace is compared against + /// `merge-base(ref, HEAD)`, so a feature branch diffs against where + /// it diverged. Two refs — `lineage --diff main feature/foo` — + /// compare the exact refs, no worktree involved. Writes + /// `lineage_diff.json` alongside the other artifacts. + #[arg(long, value_name = "BASE_REF [CANDIDATE_REF]", num_args = 1..=2, conflicts_with_all = ["target", "format"])] + diff: Vec, }, /// Show downstream impact of a column or a selection. Impact { @@ -510,14 +520,22 @@ async fn run(cli: &Cli) -> Result { Command::Apply { .. } => run_apply(cli, &compilation, &set, git.as_ref(), false).await, Command::Run { .. } => run_apply(cli, &compilation, &set, git.as_ref(), true).await, Command::Test { .. } => run_test(cli, &compilation, &set, git.as_ref()).await, - Command::Lineage { target, format } => run_lineage( - cli, - &compilation, - target.as_deref(), - &set, - git.as_ref(), - *format, - ), + Command::Lineage { + target, + format, + diff, + } => { + run_lineage( + cli, + &compilation, + target.as_deref(), + &set, + git.as_ref(), + *format, + diff.as_slice(), + ) + .await + } Command::Impact { column } => { run_impact(cli, &compilation, column.as_deref(), &set, git.as_ref()) } @@ -1209,6 +1227,19 @@ async fn run_promote( to, )?); let merge_check = nessie.can_merge(candidate, to).await.ok(); + // Lineage evidence: the artifact only speaks for this promotion when + // the identities it was produced against still hold — including the + // candidate's compiled lineage fingerprint — stale or unbound reports + // are surfaced as such, never silently as "no changes". + let current_lineage_hash = compilation.lineage.fingerprint(); + let lineage = audited_lineage( + cli, + candidate, + to, + &candidate_reference.hash, + &target.hash, + Some(¤t_lineage_hash), + ); // The `base` gate needs a target commit the evidence was established // against. Two sources, freshest first: a hash-bound diff artifact that @@ -1250,9 +1281,11 @@ async fn run_promote( "target_ref": to, "check_only": check, "gates": report.results, + "lineage": lineage, }))?; } else { print_gates_human(&report); + print_lineage_evidence(&lineage); if check && report.passed { println!("Candidate `{candidate}` can be promoted to `{to}` (check only)."); } @@ -1307,11 +1340,13 @@ async fn run_promote( print_json(&serde_json::json!({ "ok": cleanup_error.is_none(), "gates": report.results, + "lineage": lineage, "promotion": record, "cleanup_error": cleanup_error, }))?; } else { print_gates_human(&report); + print_lineage_evidence(&lineage); print_promotion_human(&record); if let Some(error) = &cleanup_error { eprintln!( @@ -1410,6 +1445,177 @@ fn contract_breaking_changes( Ok(breaking) } +/// The lineage-diff artifact's standing as evidence for this promotion. +#[derive(Clone, Debug, serde::Serialize)] +struct LineageEvidence { + /// `current` — the artifact audited this Nessie pair at these commits; + /// `advisory` — only Git-bound, so no environment identity to check; + /// `stale` — the identity it was produced for has moved and it cannot + /// be treated as describing the candidate being promoted. + status: &'static str, + /// Why the artifact is not current evidence, when stale. + #[serde(skip_serializing_if = "Option::is_none")] + reason: Option, + /// The base the diff was produced against (`ref (commit)`). + base: String, + /// Total lineage changes the artifact reports. + changes: usize, +} + +/// Read `lineage_diff.json` and audit its provenance against the pair +/// being promoted. The artifact only counts when it was produced for this +/// candidate against this target — bound to the Nessie commits when it +/// carries an environment binding, else to the Git identity it recorded — +/// and only while its candidate fingerprint still matches the compiled +/// workspace. Anything stale, unreadable, or about another pair is +/// rejected with a reason naming the rerun rather than silently treated +/// as current. +fn audited_lineage( + cli: &Cli, + candidate: &str, + to: &str, + candidate_hash: &str, + target_hash: &str, + current_lineage_hash: Option<&str>, +) -> Option { + let text = std::fs::read_to_string(artifact_path(cli, "lineage_diff.json")).ok()?; + let artifact = match serde_json::from_str::(&text) { + Ok(artifact) => artifact, + Err(error) => { + return Some(LineageEvidence { + status: "stale", + reason: Some(format!("unreadable lineage artifact: {error}")), + base: "unknown".to_string(), + changes: 0, + }); + } + }; + let base = format!( + "{} ({})", + artifact.base_ref, + short(&artifact.base_commit, 12) + ); + let changes = artifact.diff.change_count(); + let stale = |reason: String| LineageEvidence { + status: "stale", + reason: Some(reason), + base: base.clone(), + changes, + }; + let evidence = |status: &'static str| LineageEvidence { + status, + reason: None, + base: base.clone(), + changes, + }; + let rerun = format!("rerun `lineage --diff {to} --ref {candidate}`"); + let verdict = match &artifact.environment { + // Nessie-bound: the pair and both commits must match the promotion. + Some(binding) => { + if binding.candidate_ref != candidate || binding.target_ref != to { + stale(format!( + "lineage diff covers `{}` -> `{}`, not `{candidate}` -> `{to}`; {rerun}", + binding.candidate_ref, binding.target_ref + )) + } else if binding.candidate_hash != candidate_hash { + stale(format!( + "candidate `{candidate}` moved since the lineage diff; {rerun}" + )) + } else if binding.target_hash != target_hash { + stale(format!( + "target `{to}` moved since the lineage diff; {rerun}" + )) + } else { + evidence("current") + } + } + // Git-bound only: the recorded candidate identity must still hold. + None => match artifact.base_kind.as_str() { + "merge-base" => { + match phlo_transform_core::comparison_base(&cli.root, &artifact.base_ref) { + Ok(current) + if current.commit == artifact.base_commit + && current.head == artifact.candidate.head + && current.dirty == artifact.candidate.dirty => + { + evidence("advisory") + } + Ok(_) => stale(format!( + "the worktree or `{}` moved since the lineage diff; {rerun}", + artifact.base_ref + )), + Err(_) => stale(format!( + "base ref `{}` no longer resolves; {rerun}", + artifact.base_ref + )), + } + } + // Exact ref -> ref: both refs must still resolve to the commits + // the diff was produced from — a deleted or unrecorded ref is + // unverifiable, not "unchanged". + _ => { + let base_now = + match phlo_transform_core::resolve_commit(&cli.root, &artifact.base_ref) { + Ok(commit) => commit, + Err(_) => { + return Some(stale(format!( + "base ref `{}` no longer resolves; {rerun}", + artifact.base_ref + ))) + } + }; + let candidate_now = match artifact.candidate.git_ref.as_deref() { + Some(git_ref) => { + match phlo_transform_core::resolve_commit(&cli.root, git_ref) { + Ok(commit) => Some(commit), + Err(_) => { + return Some(stale(format!( + "candidate ref `{git_ref}` no longer resolves; {rerun}" + ))) + } + } + } + None => None, + }; + match (candidate_now, artifact.candidate.head.as_deref()) { + (Some(now), Some(recorded)) + if now == recorded && base_now == artifact.base_commit => + { + evidence("advisory") + } + (Some(_), Some(_)) => stale(format!( + "a diffed ref moved since the lineage diff; {rerun}" + )), + _ => stale(format!( + "the artifact does not record a resolvable candidate ref; {rerun}" + )), + } + } + }, + }; + // Whichever identity check passed, the artifact must also describe the + // candidate's current definitions: refs can sit still while edited + // code waits unpromoted, and a dirty worktree stays "dirty" as its + // contents change. + if verdict.status == "stale" { + return Some(verdict); + } + Some( + match (&artifact.candidate.lineage_hash, current_lineage_hash) { + (Some(recorded), Some(now)) if recorded == now => verdict, + (Some(_), Some(_)) => stale(format!( + "the candidate's lineage changed since the diff; {rerun}" + )), + (Some(_), None) => stale(format!( + "the candidate's lineage could not be fingerprinted; {rerun}" + )), + (None, _) => stale(format!( + "the lineage artifact predates candidate fingerprinting; {rerun}" + )), + }, + ) +} + /// Read the audited `branch_diff.json` artifact and derive the evidence it /// carries for this /// promotion: the diff verdict, why the artifact cannot be used, the breaking @@ -1640,6 +1846,26 @@ fn print_gates_human(report: &phlo_transform_engine::GateReport) { println!(); } +/// The lineage artifact's standing under the gates — advisory context, not +/// a gate itself: `stale` means the report's provenance no longer matches +/// what is being promoted. +fn print_lineage_evidence(lineage: &Option) { + let Some(lineage) = lineage else { + return; + }; + match (lineage.status, &lineage.reason) { + ("stale", Some(reason)) => println!("Lineage: stale — {reason}\n"), + ("current", _) => println!( + "Lineage: {} change(s) vs {} — current\n", + lineage.changes, lineage.base + ), + _ => println!( + "Lineage: {} change(s) vs {} (advisory — not bound to this environment)\n", + lineage.changes, lineage.base + ), + } +} + /// Removal of a promoted candidate's catalog and branch. Every failure is /// reported — the merge already happened, so a leftover catalog or branch /// must never be silent. @@ -2274,14 +2500,18 @@ async fn run_test( }) } -fn run_lineage( +async fn run_lineage( cli: &Cli, compilation: &Compilation, target: Option<&str>, set: &SelectorSet, git: Option<&GitChanges>, format: Option, + diff: &[String], ) -> Result { + if !diff.is_empty() { + return run_lineage_diff(cli, compilation, diff).await; + } // `--format` exports the canonical lineage graph: scoped to a model // target or a selection when given, the whole workspace otherwise. if let Some(format) = format { @@ -2383,6 +2613,284 @@ fn run_lineage( } } +/// `lineage --diff ` / `lineage --diff `: compile +/// the base workspace and diff its lineage graph against the candidate's. +/// +/// One ref uses the same merge-base semantics as `--since` — the workspace +/// is compared against `merge-base(base, HEAD)`, so a feature branch diffs +/// against where it diverged rather than against the other ref's current +/// head. Two refs compare the exact refs with no worktree involved. Both +/// trees are materialised read-only — the checkout is never touched. +async fn run_lineage_diff( + cli: &Cli, + compilation: &Compilation, + refs: &[String], +) -> Result { + let (base_kind, base_ref, base_tree, candidate_tree, candidate) = match refs { + [base_ref] => { + // merge-base(ref, HEAD) -> worktree: the branch-change + // workflow, identical to `--since`. + let comparison = phlo_transform_core::comparison_base(&cli.root, base_ref) + .map_err(|error| error.to_string())?; + let tree = phlo_transform_core::checkout_tree(&cli.root, &comparison.commit) + .map_err(|error| error.to_string())?; + ( + "merge-base", + base_ref.clone(), + tree, + None, + CandidateProvenance { + git_ref: environment(cli).or_else(|| cli.from.clone()), + head: comparison.head, + dirty: comparison.dirty, + lineage_hash: None, + model_versions: BTreeMap::new(), + }, + ) + } + [base_ref, candidate_ref] => { + // Exact ref -> exact ref: the candidate is a checked-out tree, + // not the working tree. + let base_tree = phlo_transform_core::checkout_tree(&cli.root, base_ref) + .map_err(|error| error.to_string())?; + let candidate_tree = phlo_transform_core::checkout_tree(&cli.root, candidate_ref) + .map_err(|error| error.to_string())?; + ( + "ref", + base_ref.clone(), + base_tree, + Some(candidate_tree), + CandidateProvenance { + git_ref: Some(candidate_ref.clone()), + head: None, + dirty: false, + lineage_hash: None, + model_versions: BTreeMap::new(), + }, + ) + } + _ => return Err("`--diff` takes one or two Git refs".to_string()), + }; + let base_commit = base_tree.commit.clone(); + let candidate = CandidateProvenance { + head: candidate_tree + .as_ref() + .map(|tree| tree.commit.clone()) + .or(candidate.head), + ..candidate + }; + + let base = match compile_tree(cli, &base_tree.workspace, &base_ref).await { + Ok(base) => base, + Err(code) => return Ok(code), + }; + let base_graph = phlo_transform_core::LineageGraph::build(&base); + let candidate_compilation = match &candidate_tree { + Some(tree) => match compile_tree(cli, &tree.workspace, &refs[1]).await { + Ok(candidate) => Some(candidate), + Err(code) => return Ok(code), + }, + None => None, + }; + let candidate_graph = candidate_compilation + .as_ref() + .map(|candidate| candidate.lineage.clone()) + .unwrap_or_else(|| compilation.lineage.clone()); + // The candidate's definitional identity: the canonical graph's + // fingerprint — the stale-artifact check promotion can trust — plus + // each model's content-addressed version as diagnostic context. + let candidate = CandidateProvenance { + lineage_hash: Some(candidate_graph.fingerprint()), + model_versions: candidate_compilation + .as_ref() + .unwrap_or(compilation) + .models + .iter() + .map(|model| (model.id.logical_name(), model.version.hash.clone())) + .collect(), + ..candidate + }; + let mut diff = phlo_transform_core::lineage_diff(&base_graph, &candidate_graph); + diff.base_ref = Some(match base_kind { + "merge-base" => format!("{base_ref} (merge-base {})", short(&base_commit, 12)), + _ => format!("{base_ref} ({})", short(&base_commit, 12)), + }); + + // Bind the diff to the Nessie pair it describes, when both resolve: + // the environment names the candidate branch, the base ref names the + // target. Unconfigured or unresolved leaves the artifact unbound — + // promotion then treats it as advisory rather than current evidence. + let environment_binding = match (candidate.git_ref.clone(), nessie_endpoint(cli)) { + (Some(candidate_ref), Some(_)) => match build_nessie(cli) { + Ok(nessie) => { + let candidate_nessie = nessie.get_reference(&candidate_ref).await.ok().flatten(); + let target_nessie = nessie.get_reference(&base_ref).await.ok().flatten(); + match (candidate_nessie, target_nessie) { + (Some(candidate), Some(target)) => Some(LineageEnvironment { + candidate_ref, + candidate_hash: candidate.hash, + target_ref: base_ref.clone(), + target_hash: target.hash, + }), + _ => None, + } + } + Err(_) => None, + }, + _ => None, + }; + + ArtifactWriter::for_workspace(&cli.root) + .write_lineage_diff(&LineageDiffArtifact { + schema_version: SCHEMA_VERSION, + base_kind: base_kind.to_string(), + base_ref: base_ref.clone(), + base_commit, + candidate, + environment: environment_binding, + diff: diff.clone(), + }) + .map_err(|error| error.to_string())?; + + if cli.json { + return print_json(&diff).map(|_| ExitCode::SUCCESS); + } + if diff.is_empty() { + println!( + "No lineage changes since {}", + diff.base_ref.as_deref().unwrap_or(&base_ref) + ); + return Ok(ExitCode::SUCCESS); + } + println!( + "Lineage diff vs {}", + diff.base_ref.as_deref().unwrap_or(&base_ref) + ); + let section = |title: &str, count: usize| { + if count > 0 { + println!(); + println!("{title}"); + } + }; + section("Added", diff.nodes_added.len()); + for node in &diff.nodes_added { + println!(" + {} ({})", node.name, node.kind); + } + section("Removed", diff.nodes_removed.len()); + for node in &diff.nodes_removed { + println!(" - {} ({})", node.name, node.kind); + } + section("Changed", diff.nodes_changed.len()); + for node in &diff.nodes_changed { + println!(" ~ {} ({})", node.name, node.kind); + for change in &node.changes { + println!(" {change}"); + } + } + if !diff.edges_added.is_empty() + || !diff.edges_removed.is_empty() + || !diff.edges_changed.is_empty() + { + println!(); + println!("Edges"); + for edge in &diff.edges_added { + let detail = edge + .detail + .as_ref() + .map(|detail| format!(" ({detail})")) + .unwrap_or_default(); + println!(" + {} --{}-> {}{detail}", edge.from, edge.kind, edge.to); + } + for edge in &diff.edges_removed { + let detail = edge + .detail + .as_ref() + .map(|detail| format!(" ({detail})")) + .unwrap_or_default(); + println!(" - {} --{}-> {}{detail}", edge.from, edge.kind, edge.to); + } + for edge in &diff.edges_changed { + println!(" ~ {} --{}-> {}", edge.from, edge.kind, edge.to); + for change in &edge.changes { + println!(" {change}"); + } + } + } + section("Impacts", diff.impacts.len() + diff.edge_impacts.len()); + for impact in &diff.impacts { + println!(" {} orphans: {}", impact.node, impact.orphans.join(", ")); + } + for impact in &diff.edge_impacts { + println!( + " {} --{}-> {} {}: affects {}", + impact.from, + impact.kind, + impact.to, + impact.change, + impact.downstream.join(", ") + ); + } + Ok(ExitCode::SUCCESS) +} + +/// Compile the workspace inside a materialised Git tree: load it, apply +/// the same `--catalog` override, and enrich it against the same adapter — +/// the live compile's twin, or every target would diff as moved. `label` +/// is the ref name diagnostics are reported under. +async fn compile_tree( + cli: &Cli, + workspace: &std::path::Path, + label: &str, +) -> Result { + let mut project = match load_project(workspace) { + Ok(project) => project, + Err(diagnostics) => { + report_diff_diagnostics(cli, label, "cannot load", &diagnostics); + return Err(ExitCode::FAILURE); + } + }; + let base_catalog = project.defaults.catalog.clone(); + let base_schema = project.defaults.schema.clone(); + if let Some(catalog) = &cli.catalog { + project.defaults.catalog = Some(catalog.clone()); + } + let compiled = { + let plain = compile(&project); + enrich( + cli, + &project, + base_catalog.as_deref(), + base_schema.as_deref(), + &plain, + ) + .await + .unwrap_or(plain) + }; + if !compiled.is_ok() { + report_diff_diagnostics( + cli, + label, + "does not compile cleanly", + &compiled.diagnostics, + ); + return Err(ExitCode::FAILURE); + } + Ok(compiled) +} + +fn report_diff_diagnostics(cli: &Cli, label: &str, problem: &str, diagnostics: &[Diagnostic]) { + if cli.json { + let _ = print_json(&serde_json::json!({ + "ok": false, + "base_ref": label, + "diagnostics": diagnostics, + })); + } else { + eprintln!("`{label}` {problem}:"); + render_diagnostics(diagnostics); + } +} + /// `lineage` with no target: the model graph as a compact edge list, /// optionally scoped to a selection. Edges are restricted to selected /// models — the listing answers "the selected subgraph", not "everything @@ -5033,4 +5541,356 @@ mod tests { "unverifiable record must block even against a keyless model: {breaks:?}" ); } + + fn lineage_artifact( + base_kind: &str, + base_ref: &str, + base_commit: &str, + candidate_head: Option<&str>, + dirty: bool, + environment: Option, + ) -> LineageDiffArtifact { + LineageDiffArtifact { + schema_version: SCHEMA_VERSION, + base_kind: base_kind.to_string(), + base_ref: base_ref.to_string(), + base_commit: base_commit.to_string(), + candidate: CandidateProvenance { + git_ref: Some("ci/x".to_string()), + head: candidate_head.map(str::to_string), + dirty, + lineage_hash: Some("fingerprint".to_string()), + model_versions: BTreeMap::new(), + }, + environment, + diff: phlo_transform_core::LineageDiff::default(), + } + } + + fn write_lineage_diff(cli: &Cli, artifact: &LineageDiffArtifact) { + ArtifactWriter::for_workspace(&cli.root) + .write_lineage_diff(artifact) + .expect("lineage artifact writes"); + } + + #[test] + fn audited_lineage_accepts_the_bound_pair() { + let dir = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(dir.path()); + write_lineage_diff( + &cli, + &lineage_artifact( + "merge-base", + "main", + "abc", + Some("def"), + false, + Some(LineageEnvironment { + candidate_ref: "ci/x".to_string(), + candidate_hash: "h1".to_string(), + target_ref: "main".to_string(), + target_hash: "h2".to_string(), + }), + ), + ); + + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "current"); + } + + /// The Nessie hashes matching is not enough: code edited after the + /// diff — or a dirty worktree edited again — changes the candidate's + /// lineage fingerprint, and the artifact stops being current. + #[test] + fn audited_lineage_rejects_a_changed_candidate_fingerprint() { + let dir = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(dir.path()); + write_lineage_diff( + &cli, + &lineage_artifact( + "merge-base", + "main", + "abc", + Some("def"), + false, + Some(LineageEnvironment { + candidate_ref: "ci/x".to_string(), + candidate_hash: "h1".to_string(), + target_ref: "main".to_string(), + target_hash: "h2".to_string(), + }), + ), + ); + + let evidence = + audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("edited")).expect("evidence"); + assert_eq!(evidence.status, "stale"); + assert!( + evidence + .reason + .as_deref() + .expect("reason") + .contains("lineage changed"), + "{:?}", + evidence.reason + ); + } + + /// An artifact written before candidate fingerprinting cannot prove it + /// describes this candidate — even when every other identity matches. + #[test] + fn audited_lineage_rejects_an_artifact_without_a_fingerprint() { + let dir = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(dir.path()); + let mut artifact = lineage_artifact( + "merge-base", + "main", + "abc", + Some("def"), + false, + Some(LineageEnvironment { + candidate_ref: "ci/x".to_string(), + candidate_hash: "h1".to_string(), + target_ref: "main".to_string(), + target_hash: "h2".to_string(), + }), + ); + artifact.candidate.lineage_hash = None; + write_lineage_diff(&cli, &artifact); + + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + assert!( + evidence + .reason + .as_deref() + .expect("reason") + .contains("fingerprinting"), + "{:?}", + evidence.reason + ); + } + + #[test] + fn audited_lineage_rejects_an_artifact_for_other_refs() { + let dir = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(dir.path()); + write_lineage_diff( + &cli, + &lineage_artifact( + "merge-base", + "main", + "abc", + Some("def"), + false, + Some(LineageEnvironment { + candidate_ref: "ci/other".to_string(), + candidate_hash: "h1".to_string(), + target_ref: "main".to_string(), + target_hash: "h2".to_string(), + }), + ), + ); + + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + assert!( + evidence + .reason + .as_deref() + .expect("reason") + .contains("ci/other"), + "{:?}", + evidence.reason + ); + } + + #[test] + fn audited_lineage_rejects_a_moved_candidate_hash() { + let dir = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(dir.path()); + write_lineage_diff( + &cli, + &lineage_artifact( + "merge-base", + "main", + "abc", + Some("def"), + false, + Some(LineageEnvironment { + candidate_ref: "ci/x".to_string(), + candidate_hash: "old".to_string(), + target_ref: "main".to_string(), + target_hash: "h2".to_string(), + }), + ), + ); + + let evidence = audited_lineage(&cli, "ci/x", "main", "new", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + assert!( + evidence + .reason + .as_deref() + .expect("reason") + .contains("moved"), + "{:?}", + evidence.reason + ); + } + + #[test] + fn audited_lineage_rejects_an_unreadable_artifact() { + let dir = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(dir.path()); + let directory = dir.path().join(".phlo").join("transform"); + std::fs::create_dir_all(&directory).expect("artifact dir"); + std::fs::write(directory.join("lineage_diff.json"), "not json").expect("write"); + + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + assert!( + evidence + .reason + .as_deref() + .expect("reason") + .contains("unreadable"), + "{:?}", + evidence.reason + ); + } + + #[test] + fn audited_lineage_absent_artifact_is_none() { + let dir = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(dir.path()); + assert!(audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")).is_none()); + } + + /// A git-bound (no Nessie binding) merge-base artifact is advisory while + /// HEAD and the merge-base still match, and stale once the worktree or + /// base moved. A repo the artifact never knew cannot prove currentness. + #[test] + fn audited_lineage_git_bound_tracks_the_worktree() { + let dir = tempfile::tempdir().expect("tempdir"); + let root = dir.path(); + let git = |args: &[&str]| { + let output = std::process::Command::new("git") + .arg("-C") + .arg(root) + .args(args) + .output() + .expect("git runs"); + assert!(output.status.success(), "{args:?}"); + String::from_utf8_lossy(&output.stdout).trim().to_string() + }; + git(&["init", "-q", "-b", "main"]); + git(&["config", "user.email", "test@phlo.dev"]); + git(&["config", "user.name", "Phlo Test"]); + std::fs::write(root.join("phlo.toml"), "[transform]\nroots = []\n").expect("toml"); + git(&["add", "-A"]); + git(&["commit", "-qm", "init"]); + let head = git(&["rev-parse", "HEAD"]); + + let cli = cli_at(root); + // Current merge-base, current head, clean — advisory. + write_lineage_diff( + &cli, + &lineage_artifact("merge-base", "main", &head, Some(&head), false, None), + ); + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "advisory"); + + // HEAD recorded differently — the worktree moved on. + write_lineage_diff( + &cli, + &lineage_artifact("merge-base", "main", &head, Some("stale"), false, None), + ); + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + + // The same artifact in a non-repository cannot resolve its base. + let bare = tempfile::tempdir().expect("tempdir"); + let cli = cli_at(bare.path()); + write_lineage_diff( + &cli, + &lineage_artifact("merge-base", "main", &head, Some(&head), false, None), + ); + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + } + + /// An exact ref→ref artifact is only evidence while both refs still + /// resolve to the commits it recorded — deleting either ref is + /// unverifiable, not "unchanged". + #[test] + fn audited_lineage_rejects_deleted_refs() { + let dir = tempfile::tempdir().expect("tempdir"); + let root = dir.path(); + let git = |args: &[&str]| { + let output = std::process::Command::new("git") + .arg("-C") + .arg(root) + .args(args) + .output() + .expect("git runs"); + assert!(output.status.success(), "{args:?}"); + String::from_utf8_lossy(&output.stdout).trim().to_string() + }; + git(&["init", "-q", "-b", "main"]); + git(&["config", "user.email", "test@phlo.dev"]); + git(&["config", "user.name", "Phlo Test"]); + std::fs::write(root.join("phlo.toml"), "[transform]\nroots = []\n").expect("toml"); + git(&["add", "-A"]); + git(&["commit", "-qm", "init"]); + let head = git(&["rev-parse", "HEAD"]); + git(&["branch", "base"]); + git(&["branch", "feature"]); + + let cli = cli_at(root); + let mut artifact = lineage_artifact("ref", "base", &head, Some(&head), false, None); + artifact.candidate.git_ref = Some("feature".to_string()); + write_lineage_diff(&cli, &artifact); + + // Both refs resolve to the recorded commits — advisory. + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "advisory"); + + // The candidate ref is gone — the artifact can no longer prove it + // describes `feature`. + git(&["branch", "-D", "feature"]); + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + assert!( + evidence + .reason + .as_deref() + .expect("reason") + .contains("feature"), + "{:?}", + evidence.reason + ); + + // Restore the candidate; deleting the base ref must go stale too. + git(&["branch", "feature"]); + git(&["branch", "-D", "base"]); + let evidence = audited_lineage(&cli, "ci/x", "main", "h1", "h2", Some("fingerprint")) + .expect("evidence"); + assert_eq!(evidence.status, "stale"); + assert!( + evidence.reason.as_deref().expect("reason").contains("base"), + "{:?}", + evidence.reason + ); + } } diff --git a/crates/phlo-transform-cli/tests/cli.rs b/crates/phlo-transform-cli/tests/cli.rs index 1032276..f35b186 100644 --- a/crates/phlo-transform-cli/tests/cli.rs +++ b/crates/phlo-transform-cli/tests/cli.rs @@ -1033,6 +1033,155 @@ fn plan_since_marks_modified_model() { assert!(!body.contains("BUILD assay.results"), "{body}"); } +/// `lineage --diff` on a clean repo reports no changes. +#[test] +fn lineage_diff_clean_repo_reports_nothing() { + let dir = git_workspace(); + let output = plan_since(dir.path(), &["lineage", "--diff", "main"]); + assert!(output.status.success(), "{}", stdout(&output)); + assert!( + stdout(&output).contains("No lineage changes"), + "{}", + stdout(&output) + ); +} + +/// A deleted model shows as removed and names the consumer it orphans. +#[test] +fn lineage_diff_reports_removed_model_and_orphans() { + let dir = git_workspace(); + std::fs::remove_file(dir.path().join("transforms/assay/raw.sql")).expect("delete"); + let output = plan_since(dir.path(), &["lineage", "--diff", "main"]); + assert!(output.status.success(), "{}", stdout(&output)); + let body = stdout(&output); + assert!(body.contains("- assay.raw (model)"), "{body}"); + assert!(body.contains("orphans"), "{body}"); + assert!(body.contains("assay.results"), "{body}"); +} + +/// An edited model shows as changed with its version delta. +#[test] +fn lineage_diff_reports_changed_model() { + let dir = git_workspace(); + std::fs::write( + dir.path().join("transforms/assay/results.sql"), + "select id, id + 1 as extra from assay.raw\n", + ) + .expect("edit"); + let output = plan_since(dir.path(), &["lineage", "--diff", "main", "--json"]); + assert!(output.status.success(), "{}", stdout(&output)); + let body = stdout(&output); + let json: serde_json::Value = serde_json::from_str(&body).expect("json"); + let changed = json["nodes_changed"].as_array().expect("nodes_changed"); + assert!( + changed.iter().any(|node| node["name"] == "assay.results"), + "{body}" + ); +} + +/// `lineage --diff` writes the `lineage_diff.json` artifact. +#[test] +fn lineage_diff_writes_artifact() { + let dir = git_workspace(); + let output = plan_since(dir.path(), &["lineage", "--diff", "main"]); + assert!(output.status.success(), "{}", stdout(&output)); + let artifact = dir.path().join(".phlo/transform/lineage_diff.json"); + let text = std::fs::read_to_string(&artifact).expect("artifact written"); + let json: serde_json::Value = serde_json::from_str(&text).expect("artifact json"); + assert_eq!(json["base_ref"], "main"); + assert!(json["base_commit"].is_string(), "{text}"); + // Provenance: the base is the merge-base, and the candidate records the + // git head + worktree state the diff was produced against. + assert_eq!(json["base_kind"], "merge-base", "{text}"); + let head = String::from_utf8_lossy( + &std::process::Command::new("git") + .arg("-C") + .arg(dir.path()) + .args(["rev-parse", "HEAD"]) + .output() + .expect("git rev-parse") + .stdout, + ) + .trim() + .to_string(); + assert_eq!(json["candidate"]["head"], head, "{text}"); + assert_eq!(json["candidate"]["dirty"], false, "{text}"); + // The candidate's lineage fingerprint binds the artifact to these + // definitions — promotion audits it against the compiled workspace. + assert!(json["candidate"]["lineage_hash"].is_string(), "{text}"); + assert!( + json["candidate"]["model_versions"]["assay.raw"].is_string(), + "{text}" + ); +} + +/// `lineage --diff main` uses merge-base semantics: work committed on +/// `main` after the branch diverged is not part of the comparison. +#[test] +fn lineage_diff_uses_merge_base_not_ref_head() { + let dir = git_workspace(); + let root = dir.path(); + // A ── feature; then main advances to B with a model the feature + // never had. + git(root, &["checkout", "-q", "-b", "feature"]); + git(root, &["checkout", "-q", "main"]); + std::fs::write( + root.join("transforms/assay/main_only.sql"), + "select 1 as id\n", + ) + .expect("main-only model"); + git(root, &["add", "-A"]); + git(root, &["commit", "-qm", "main moves on"]); + git(root, &["checkout", "-q", "feature"]); + // The feature's own change, uncommitted in the worktree. + std::fs::write( + root.join("transforms/assay/results.sql"), + "select id, id + 1 as extra from assay.raw\n", + ) + .expect("edit"); + + let output = plan_since(root, &["lineage", "--diff", "main"]); + assert!(output.status.success(), "{}", stdout(&output)); + let body = stdout(&output); + // merge-base(main, feature) = A: `main_only` is on neither side — with + // direct `main^{commit}` semantics it would report as removed. + assert!(!body.contains("main_only"), "{body}"); + // The feature's real change still reports, labelled as a merge-base diff. + assert!(body.contains("assay.results"), "{body}"); + assert!(body.contains("merge-base"), "{body}"); +} + +/// `--diff base candidate` compares the exact refs — the dirty worktree +/// is not part of either side. +#[test] +fn lineage_diff_two_refs_compares_exactly() { + let dir = git_workspace(); + let root = dir.path(); + git(root, &["checkout", "-q", "-b", "feature"]); + std::fs::write( + root.join("transforms/assay/feature_only.sql"), + "select 1 as id\n", + ) + .expect("feature model"); + git(root, &["add", "-A"]); + git(root, &["commit", "-qm", "feature work"]); + // An uncommitted model — invisible to an exact ref comparison. + std::fs::write(root.join("transforms/assay/dirty.sql"), "select 1 as id\n") + .expect("uncommitted model"); + + let output = plan_since(root, &["lineage", "--diff", "main", "feature"]); + assert!(output.status.success(), "{}", stdout(&output)); + let body = stdout(&output); + assert!(body.contains("+ assay.feature_only (model)"), "{body}"); + assert!(!body.contains("dirty"), "{body}"); + // Exact refs name the resolved head, not a merge-base. + let artifact = root.join(".phlo/transform/lineage_diff.json"); + let text = std::fs::read_to_string(&artifact).expect("artifact written"); + let json: serde_json::Value = serde_json::from_str(&text).expect("artifact json"); + assert_eq!(json["base_kind"], "ref", "{text}"); + assert_eq!(json["candidate"]["git_ref"], "feature", "{text}"); +} + /// `changed+` expands the Git change set downstream through the graph. #[test] fn plan_since_changed_plus_expands_downstream() { diff --git a/crates/phlo-transform-core/Cargo.toml b/crates/phlo-transform-core/Cargo.toml index a567313..788f17d 100644 --- a/crates/phlo-transform-core/Cargo.toml +++ b/crates/phlo-transform-core/Cargo.toml @@ -11,8 +11,10 @@ rust-version.workspace = true globset = { workspace = true } petgraph = { workspace = true } serde = { workspace = true } +serde_json = { workspace = true } sha2 = { workspace = true } sqlparser = { workspace = true } +tempfile = "3" thiserror = { workspace = true } toml = { workspace = true } walkdir = { workspace = true } @@ -20,5 +22,3 @@ phlo-transform-sql = { workspace = true } [dev-dependencies] insta = { version = "1", features = ["json"] } -serde_json = { workspace = true } -tempfile = "3" diff --git a/crates/phlo-transform-core/src/git.rs b/crates/phlo-transform-core/src/git.rs index 5e5c681..212668a 100644 --- a/crates/phlo-transform-core/src/git.rs +++ b/crates/phlo-transform-core/src/git.rs @@ -193,6 +193,227 @@ pub enum GitError { Command(String), } +/// A workspace tree materialised at one commit — every file under the +/// workspace prefix at `commit`, extracted into a temporary directory so +/// the base side can be discovered and compiled exactly as a checkout +/// would be. +pub struct CheckedTree { + /// The extraction — kept alive so `workspace` stays valid; dropped + /// (and deleted) with the value. + pub dir: tempfile::TempDir, + /// The extracted workspace root: `/`. + pub workspace: std::path::PathBuf, + /// The commit the tree was extracted from. + pub commit: String, +} + +/// What a workspace diff compares against: the merge-base `git_ref` shares +/// with HEAD — the same baseline `--since` uses — plus the candidate side's +/// identity. +#[derive(Clone, Debug)] +pub struct ComparisonBase { + /// The merge-base commit (the empty tree when HEAD has no commits). + pub commit: String, + /// HEAD's commit at resolution time, when the checkout has one. + pub head: Option, + /// Whether the worktree carries uncommitted or untracked changes under + /// the workspace root — the candidate side of a workspace diff is the + /// working tree, so dirt is part of its identity. + pub dirty: bool, +} + +/// Resolve the branch-diff baseline for `git_ref`: `merge-base(git_ref, +/// HEAD)`, so a feature branch is compared against where it diverged — +/// not against the other ref's current head, which may have moved on. +pub fn comparison_base(workspace_root: &Path, git_ref: &str) -> Result { + let workspace = std::fs::canonicalize(workspace_root).map_err(|error| { + GitError::Command(format!( + "could not resolve workspace root `{}`: {error}", + workspace_root.display() + )) + })?; + let repo_text = git_stdout(&workspace, &["rev-parse", "--show-toplevel"]) + .ok_or_else(|| GitError::NotARepository(workspace.display().to_string()))?; + let repo = std::fs::canonicalize(repo_text.trim()) + .map_err(|error| GitError::Command(format!("could not resolve repo root: {error}")))?; + let prefix = workspace + .strip_prefix(&repo) + .unwrap_or(&workspace) + .to_path_buf(); + + let reference = git_stdout( + &repo, + &[ + "rev-parse", + "--verify", + "--quiet", + &format!("{git_ref}^{{commit}}"), + ], + ) + .ok_or_else(|| GitError::UnknownRef(git_ref.to_string()))?; + let head = git_stdout(&repo, &["rev-parse", "--verify", "--quiet", "HEAD"]) + .map(|head| head.trim().to_string()); + let commit = match &head { + Some(head) => git_stdout(&repo, &["merge-base", reference.trim(), head.trim()]) + .map(|base| base.trim().to_string()) + .ok_or_else(|| GitError::NoMergeBase { + since: git_ref.to_string(), + })?, + None => EMPTY_TREE.to_string(), + }; + let spec = if prefix.as_os_str().is_empty() { + ".".to_string() + } else { + prefix.to_string_lossy().replace('\\', "/") + }; + // `.phlo/` holds artifacts — writing `lineage_diff.json` must not make + // the worktree it describes "dirty". + let exclude = if prefix.as_os_str().is_empty() { + ":(exclude).phlo".to_string() + } else { + format!(":(exclude){}/.phlo", spec) + }; + let dirty = git_stdout(&repo, &["status", "--porcelain", "--", &spec, &exclude]) + .map(|status| !status.trim().is_empty()) + .unwrap_or(false); + Ok(ComparisonBase { + commit, + head, + dirty, + }) +} + +/// The commit `git_ref` currently resolves to in the workspace's +/// repository. +pub fn resolve_commit(workspace_root: &Path, git_ref: &str) -> Result { + let workspace = std::fs::canonicalize(workspace_root).map_err(|error| { + GitError::Command(format!( + "could not resolve workspace root `{}`: {error}", + workspace_root.display() + )) + })?; + let repo_text = git_stdout(&workspace, &["rev-parse", "--show-toplevel"]) + .ok_or_else(|| GitError::NotARepository(workspace.display().to_string()))?; + git_stdout( + Path::new(repo_text.trim()), + &[ + "rev-parse", + "--verify", + "--quiet", + &format!("{git_ref}^{{commit}}"), + ], + ) + .map(|commit| commit.trim().to_string()) + .ok_or_else(|| GitError::UnknownRef(git_ref.to_string())) +} + +/// Materialise the workspace at `git_ref` into a temporary directory. +/// +/// `git ls-tree` lists the workspace subtree at the resolved commit and one +/// `cat-file --batch` round fetches every blob — the repository is never +/// mutated (no worktree, no checkout), so this is safe to run alongside an +/// in-progress rebase or a dirty working tree. +pub fn checkout_tree(workspace_root: &Path, git_ref: &str) -> Result { + let workspace = std::fs::canonicalize(workspace_root).map_err(|error| { + GitError::Command(format!( + "could not resolve workspace root `{}`: {error}", + workspace_root.display() + )) + })?; + let repo_text = git_stdout(&workspace, &["rev-parse", "--show-toplevel"]) + .ok_or_else(|| GitError::NotARepository(workspace.display().to_string()))?; + let repo = std::fs::canonicalize(repo_text.trim()) + .map_err(|error| GitError::Command(format!("could not resolve repo root: {error}")))?; + let prefix = workspace + .strip_prefix(&repo) + .unwrap_or(&workspace) + .to_path_buf(); + + let commit = git_stdout( + &repo, + &[ + "rev-parse", + "--verify", + "--quiet", + &format!("{git_ref}^{{commit}}"), + ], + ) + .ok_or_else(|| GitError::UnknownRef(git_ref.to_string()))?; + let commit = commit.trim().to_string(); + + // `ls-tree -r -z -- `: ` \t` + // entries, NUL-separated. Only blobs become files — gitlinks + // (submodule commit entries) are skipped. + let spec = if prefix.as_os_str().is_empty() { + ".".to_string() + } else { + prefix.to_string_lossy().replace('\\', "/") + }; + let listing = git_bytes(&repo, &["ls-tree", "-r", "-z", &commit, "--", &spec])?; + let mut blobs = Vec::new(); + let mut symlinks = Vec::new(); + for entry in listing.split(|byte| *byte == 0).filter(|e| !e.is_empty()) { + let entry = String::from_utf8_lossy(entry); + let Some((meta, path)) = entry.split_once('\t') else { + continue; + }; + let mut parts = meta.split_whitespace(); + let (mode, kind) = (parts.next().unwrap_or(""), parts.next().unwrap_or("")); + if kind != "blob" { + continue; + } + if mode == "120000" { + symlinks.push(path.to_string()); + } else { + blobs.push(path.to_string()); + } + } + + let dir = tempfile::tempdir() + .map_err(|error| GitError::Command(format!("could not create tempdir: {error}")))?; + let contents = blob_contents(&repo, &commit, &blobs)?; + for (path, content) in &contents { + let Some(content) = content else { + continue; + }; + let target = dir.path().join(path); + if let Some(parent) = target.parent() { + std::fs::create_dir_all(parent) + .map_err(|error| GitError::Command(format!("could not extract {path}: {error}")))?; + } + std::fs::write(&target, content) + .map_err(|error| GitError::Command(format!("could not extract {path}: {error}")))?; + } + // Symlinked files: recreate the link so the extraction reads what the + // checkout would. On platforms without symlink support the target + // path text is written instead — the file still exists for discovery. + let links = blob_contents(&repo, &commit, &symlinks)?; + for (path, link_target) in &links { + let Some(link_target) = link_target else { + continue; + }; + let link = dir.path().join(path); + if let Some(parent) = link.parent() { + let _ = std::fs::create_dir_all(parent); + } + #[cfg(unix)] + { + let _ = std::os::unix::fs::symlink(link_target.trim_end(), &link); + } + #[cfg(not(unix))] + { + let _ = std::fs::write(&link, link_target); + } + } + + let workspace = dir.path().join(&prefix); + Ok(CheckedTree { + dir, + workspace, + commit, + }) +} + /// What a workspace-relative path is to the project. #[derive(Clone, Copy, PartialEq, Eq)] enum PathKind { diff --git a/crates/phlo-transform-core/src/lib.rs b/crates/phlo-transform-core/src/lib.rs index 4d27119..9ef8e4d 100644 --- a/crates/phlo-transform-core/src/lib.rs +++ b/crates/phlo-transform-core/src/lib.rs @@ -24,6 +24,7 @@ pub mod git; pub mod graph; pub mod identity; pub mod lineage; +pub mod lineage_diff; pub mod model; pub mod report; pub mod resolve; @@ -42,8 +43,9 @@ pub use consumers::{ConsumerRegistry, EmptyConsumerRegistry, StaticConsumerRegis pub use diagnostics::{codes, Diagnostic, Severity}; pub use discovery::{load_project, model_id_for_path}; pub use git::{ - changes as git_changes, ChangedModel, ChangedPath, ChangedSeed, ChangedTest, DeletedModel, - GitChanges, GitError, PathStatus, + changes as git_changes, checkout_tree, comparison_base, resolve_commit, ChangedModel, + ChangedPath, ChangedSeed, ChangedTest, CheckedTree, ComparisonBase, DeletedModel, GitChanges, + GitError, PathStatus, }; pub use graph::{Dependency, EdgeKind, GraphNode, TransformGraph}; pub use identity::{IdentityError, ModelId, Namespace, SourceId}; @@ -51,6 +53,9 @@ pub use lineage::{ DatasetColumn, DatasetId, DatasetKind, LineageDocument, LineageEdge, LineageEdgeKind, LineageGraph, LineageNode, NodeMeta, }; +pub use lineage_diff::{ + lineage_diff, DiffEdge, DiffNode, EdgeChange, LineageDiff, LineageDiffImpact, NodeChange, +}; pub use model::{ FrontendKind, ModelConfig, ModelOrigin, Relation, RootKind, RootNamespaceStrategy, RootRef, SemanticModel, SemanticProject, SemanticSeed, SemanticTest, TestId, TransformRoot, diff --git a/crates/phlo-transform-core/src/lineage.rs b/crates/phlo-transform-core/src/lineage.rs index 3b422bc..b0ff1e7 100644 --- a/crates/phlo-transform-core/src/lineage.rs +++ b/crates/phlo-transform-core/src/lineage.rs @@ -175,7 +175,7 @@ impl LineageNode { } /// Optional metadata carried by a node. -#[derive(Clone, Debug, Default, Serialize)] +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] pub struct NodeMeta { /// Workspace-relative source path, when the node is file-backed. #[serde(skip_serializing_if = "Option::is_none")] @@ -726,6 +726,38 @@ impl LineageGraph { LineageDocument { nodes, edges } } + /// A content fingerprint of the whole graph — nodes, edges and their + /// metadata in canonical order, so two compilations that produce the + /// same lineage hash identically. Promotion uses it to prove a lineage + /// diff artifact still describes the candidate being promoted. + pub fn fingerprint(&self) -> String { + let mut document = self.document(); + // The document's edge order is only (from, to)-canonical; parallel + // edges need their metadata in the sort key for a total order. + document.edges.sort_by(|left, right| { + ( + &left.from, + &left.to, + &left.edge.kind, + &left.edge.directness, + &left.edge.transformation, + &left.edge.confidence, + &left.edge.expression, + ) + .cmp(&( + &right.from, + &right.to, + &right.edge.kind, + &right.edge.directness, + &right.edge.transformation, + &right.edge.confidence, + &right.edge.expression, + )) + }); + let canonical = serde_json::to_string(&document).expect("a lineage document serialises"); + crate::version::sha256_hex(&canonical) + } + /// A document restricted to a selection of models: the models, their /// output datasets and columns, the datasets they read, and the tests /// asserting on them. Edges are restricted to those in-scope nodes — diff --git a/crates/phlo-transform-core/src/lineage_diff.rs b/crates/phlo-transform-core/src/lineage_diff.rs new file mode 100644 index 0000000..cf2a820 --- /dev/null +++ b/crates/phlo-transform-core/src/lineage_diff.rs @@ -0,0 +1,841 @@ +//! Semantic diff between two lineage graphs. +//! +//! `lineage_diff` compares the base graph (what another ref or environment +//! compiled to) against the candidate graph (the current workspace) and +//! reports the structural delta: nodes added, removed or metadata-changed, +//! dependency edges gained or lost, and column-lineage edges whose +//! transformation metadata moved. Removed nodes carry the base-side +//! downstream consumers they orphan — the semantic half of impact +//! analysis, complementing the data-side `branch_diff` report. +//! +//! The report is plain serialisable data — the CLI prints it, the +//! `lineage_diff.json` artifact persists it, and agent tooling consumes +//! the same shape. + +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +use crate::lineage::{LineageEdge, LineageEdgeKind, LineageGraph, LineageNode, NodeMeta}; + +/// One node in the diff — identified by URI, labelled by kind and name. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct DiffNode { + pub uri: String, + pub kind: String, + pub name: String, +} + +/// A node present on both sides whose metadata changed. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct NodeChange { + pub uri: String, + pub kind: String, + pub name: String, + /// `field: old -> new` descriptions, in field order. + pub changes: Vec, +} + +/// One edge in the diff — identity is `(from, to, kind)` plus, for edges +/// carrying metadata, a `detail` summary distinguishing parallel edges. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct DiffEdge { + pub from: String, + pub to: String, + pub kind: String, + /// Directness/transformation/expression summary — present only on edges + /// that carry metadata (column `derives` edges), so two parallel edges + /// between the same nodes remain distinguishable. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub detail: Option, +} + +/// An edge present on both sides whose recorded metadata changed — +/// a column-level `derives` edge whose transformation, directness, +/// confidence or expression moved. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct EdgeChange { + pub from: String, + pub to: String, + pub kind: String, + /// `field: old -> new` descriptions. + pub changes: Vec, +} + +/// What a removed node leaves behind: the nodes that consumed it on the +/// base side and are now reading a dataset or model that no longer exists. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LineageDiffImpact { + /// URI of the removed node. + pub node: String, + /// Dotted names of its transitive downstream consumers on the base + /// side (models, datasets and tests — not the node's own columns). + pub orphans: Vec, +} + +/// A lineage path that disappeared or moved: `to` lost — or changed — its +/// `from` input on the base side, and everything downstream of `to` loses +/// or alters that contribution. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct EdgeImpact { + pub from: String, + pub to: String, + pub kind: String, + /// `removed` when the edge is gone entirely, `changed` when only its + /// metadata (directness, transformation, expression) moved. + pub change: String, + /// Dotted names of `to`'s transitive downstream consumers on the base + /// side — the nodes whose data no longer includes this path, or + /// includes it altered. + pub downstream: Vec, +} + +/// The structural delta between two lineage graphs. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct LineageDiff { + /// Label for the side the diff compares against (a git ref, `main`, + /// a commit). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_ref: Option, + /// Nodes only in the candidate graph. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub nodes_added: Vec, + /// Nodes only in the base graph. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub nodes_removed: Vec, + /// Nodes on both sides whose metadata changed. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub nodes_changed: Vec, + /// Edges only in the candidate graph. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub edges_added: Vec, + /// Edges only in the base graph. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub edges_removed: Vec, + /// Edges on both sides whose metadata changed. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub edges_changed: Vec, + /// Downstream consumers orphaned by each removed node. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub impacts: Vec, + /// Lineage paths that disappeared or changed — each removed or + /// metadata-moved edge and the base-side downstream it fed. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub edge_impacts: Vec, +} + +impl LineageDiff { + /// Whether the diff reports any change at all. + pub fn is_empty(&self) -> bool { + self.nodes_added.is_empty() + && self.nodes_removed.is_empty() + && self.nodes_changed.is_empty() + && self.edges_added.is_empty() + && self.edges_removed.is_empty() + && self.edges_changed.is_empty() + } + + /// Total count of changed entries — for one-line summaries. + pub fn change_count(&self) -> usize { + self.nodes_added.len() + + self.nodes_removed.len() + + self.nodes_changed.len() + + self.edges_added.len() + + self.edges_removed.len() + + self.edges_changed.len() + } +} + +/// Diff `candidate` against `base`. Deterministic: both graphs iterate in +/// sorted order and every report list is sorted by its identity fields. +pub fn lineage_diff(base: &LineageGraph, candidate: &LineageGraph) -> LineageDiff { + let mut diff = LineageDiff::default(); + + // --- Nodes --------------------------------------------------------- + let base_nodes: BTreeMap = + base.nodes().map(|(n, m)| (n.clone(), m.clone())).collect(); + let candidate_nodes: BTreeMap = candidate + .nodes() + .map(|(n, m)| (n.clone(), m.clone())) + .collect(); + + for (node, meta) in &candidate_nodes { + match base_nodes.get(node) { + None => diff.nodes_added.push(diff_node(node)), + Some(base_meta) => { + let changes = meta_changes(base_meta, meta); + if !changes.is_empty() { + let entry = diff_node(node); + diff.nodes_changed.push(NodeChange { + uri: entry.uri, + kind: entry.kind, + name: entry.name, + changes, + }); + } + } + } + } + for node in base_nodes.keys() { + if !candidate_nodes.contains_key(node) { + diff.nodes_removed.push(diff_node(node)); + } + } + + // --- Edges --------------------------------------------------------- + // Edges are a multiset, not a map: `LineageGraph` permits parallel + // edges between the same nodes when their metadata differs (two column + // `derives` edges with different directness or expressions). Keying by + // `(from, to, kind)` alone would collapse them. + type EdgeKey = (LineageNode, LineageNode, LineageEdgeKind); + let edge_map = |graph: &LineageGraph| -> BTreeMap> { + let mut map: BTreeMap> = BTreeMap::new(); + for (from, to, edge) in graph.edges() { + map.entry((from.clone(), to.clone(), edge.kind)) + .or_default() + .push(edge.clone()); + } + // Sorted bags make the pairwise comparison deterministic. + for edges in map.values_mut() { + edges.sort_by(|left, right| edge_sort_key(left).cmp(&edge_sort_key(right))); + } + map + }; + let base_edges = edge_map(base); + let candidate_edges = edge_map(candidate); + + for (key, candidate_bag) in &candidate_edges { + let (from, to, kind) = key; + let Some(base_bag) = base_edges.get(key) else { + for edge in candidate_bag { + diff.edges_added.push(diff_edge(from, to, *kind, edge)); + } + continue; + }; + // Cancel identical edges pairwise; leftovers on each side are then + // paired positionally as metadata changes, with any excess reported + // as added/removed. + let mut base_left: Vec<&LineageEdge> = Vec::new(); + let mut candidate_left: Vec<&LineageEdge> = Vec::new(); + let mut base_iter = base_bag.iter().peekable(); + let mut candidate_iter = candidate_bag.iter().peekable(); + loop { + match (base_iter.peek(), candidate_iter.peek()) { + (Some(base_edge), Some(candidate_edge)) if base_edge == candidate_edge => { + base_iter.next(); + candidate_iter.next(); + } + (Some(base_edge), Some(candidate_edge)) => { + // Both are sorted by metadata — drop the smaller side. + if edge_sort_key(base_edge) <= edge_sort_key(candidate_edge) { + base_left.push(base_iter.next().expect("peeked")); + } else { + candidate_left.push(candidate_iter.next().expect("peeked")); + } + } + (Some(_), None) => base_left.push(base_iter.next().expect("peeked")), + (None, Some(_)) => candidate_left.push(candidate_iter.next().expect("peeked")), + (None, None) => break, + } + } + // Equal leftovers pair as metadata changes; an excess reports as + // genuinely added or removed edges. + for (base_edge, candidate_edge) in base_left.iter().zip(candidate_left.iter()) { + let changes = edge_changes(base_edge, candidate_edge); + debug_assert!( + !changes.is_empty(), + "leftover edges always differ in metadata" + ); + diff.edges_changed.push(EdgeChange { + from: from.uri(), + to: to.uri(), + kind: edge_kind_label(*kind).to_string(), + changes, + }); + } + for edge in base_left.iter().skip(candidate_left.len()) { + diff.edges_removed.push(diff_edge(from, to, *kind, edge)); + } + for edge in candidate_left.iter().skip(base_left.len()) { + diff.edges_added.push(diff_edge(from, to, *kind, edge)); + } + } + for (key, base_bag) in &base_edges { + if !candidate_edges.contains_key(key) { + let (from, to, kind) = key; + for edge in base_bag { + diff.edges_removed.push(diff_edge(from, to, *kind, edge)); + } + } + } + + // --- Impacts --------------------------------------------------------- + // A removed node's orphans are its base-side downstream consumers — + // the things that were reading it and now face a gap. + for node in &diff.nodes_removed { + let base_node = node_key(base, &node.uri); + let Some(base_node) = base_node else { + continue; + }; + // Model and output dataset share a logical name — dedupe so a + // consumer isn't listed twice. + let mut seen = std::collections::BTreeSet::new(); + let orphans: Vec = base + .downstream_transitive(&base_node) + .into_iter() + .filter_map(|node| node_name(&node)) + .filter(|name| seen.insert(name.clone())) + .collect(); + if !orphans.is_empty() { + diff.impacts.push(LineageDiffImpact { + node: node.uri.clone(), + orphans, + }); + } + } + + // A removed or metadata-moved edge is a lineage path that disappeared + // or changed: everything downstream of the edge's consumer loses — or + // alters — that contribution. Base-side traversal only: the path no + // longer exists on the candidate. + let mut moved: Vec<(String, String, String, &str)> = Vec::new(); + for edge in &diff.edges_removed { + moved.push(( + edge.from.clone(), + edge.to.clone(), + edge.kind.clone(), + "removed", + )); + } + for edge in &diff.edges_changed { + moved.push(( + edge.from.clone(), + edge.to.clone(), + edge.kind.clone(), + "changed", + )); + } + for (from, to, kind, change) in moved { + let Some(to_node) = node_key(base, &to) else { + continue; + }; + let mut seen = std::collections::BTreeSet::new(); + let downstream: Vec = base + .downstream_transitive(&to_node) + .into_iter() + .filter_map(|node| node_name(&node)) + .filter(|name| seen.insert(name.clone())) + .collect(); + if !downstream.is_empty() { + diff.edge_impacts.push(EdgeImpact { + from, + to, + kind, + change: change.to_string(), + downstream, + }); + } + } + + diff +} + +fn diff_node(node: &LineageNode) -> DiffNode { + DiffNode { + uri: node.uri(), + kind: node.kind().to_string(), + name: node_name(node).unwrap_or_else(|| node.uri()), + } +} + +/// The human label for a node — logical names where they exist. +fn node_name(node: &LineageNode) -> Option { + match node { + LineageNode::Model(id) => Some(id.logical_name()), + LineageNode::Dataset(id) => Some(id.name()), + LineageNode::Column(column) => Some(format!("{}.{}", column.dataset.name(), column.name)), + LineageNode::Test(id) => Some(id.name().to_string()), + } +} + +fn node_key(graph: &LineageGraph, uri: &str) -> Option { + graph + .nodes() + .map(|(node, _)| node.clone()) + .find(|node| node.uri() == uri) +} + +fn diff_edge( + from: &LineageNode, + to: &LineageNode, + kind: LineageEdgeKind, + edge: &LineageEdge, +) -> DiffEdge { + DiffEdge { + from: from.uri(), + to: to.uri(), + kind: edge_kind_label(kind).to_string(), + detail: edge_detail(edge), + } +} + +/// Total order over an edge's metadata — the bag-sort key for the multiset +/// diff. Identity fields (`from`, `to`, `kind`) are the map key; this orders +/// within it. +fn edge_sort_key( + edge: &LineageEdge, +) -> ( + Option, + Option, + Option, + Option<&str>, +) { + ( + edge.directness, + edge.transformation, + edge.confidence, + edge.expression.as_deref(), + ) +} + +/// A one-line summary of an edge's metadata, so parallel edges between the +/// same two nodes stay distinguishable in the report. +fn edge_detail(edge: &LineageEdge) -> Option { + let mut parts = Vec::new(); + if let Some(directness) = &edge.directness { + parts.push(format!("{directness:?}").to_lowercase()); + } + if let Some(transformation) = &edge.transformation { + parts.push(format!("{transformation:?}").to_lowercase()); + } + if let Some(expression) = &edge.expression { + parts.push(format!("`{}`", truncate(expression, 60))); + } + (!parts.is_empty()).then(|| parts.join(", ")) +} + +fn edge_kind_label(kind: LineageEdgeKind) -> &'static str { + match kind { + LineageEdgeKind::Input => "input", + LineageEdgeKind::Output => "output", + LineageEdgeKind::Derives => "derives", + LineageEdgeKind::Contains => "contains", + LineageEdgeKind::Tests => "tests", + } +} + +fn opt(value: &Option) -> &str { + value.as_deref().unwrap_or("-") +} + +/// `field: old -> new` descriptions for every NodeMeta field that differs. +fn meta_changes(base: &NodeMeta, candidate: &NodeMeta) -> Vec { + let mut changes = Vec::new(); + if base.path != candidate.path { + changes.push(format!( + "path: {} -> {}", + opt(&base.path), + opt(&candidate.path) + )); + } + if base.version != candidate.version { + // Versions are content hashes — shorten for readability. + changes.push(format!( + "version: {} -> {}", + short(opt(&base.version)), + short(opt(&candidate.version)) + )); + } + if base.dataset_kind != candidate.dataset_kind { + changes.push(format!( + "dataset kind: {} -> {}", + dataset_kind_label(base.dataset_kind), + dataset_kind_label(candidate.dataset_kind) + )); + } + if base.target != candidate.target { + changes.push(format!( + "target: {} -> {}", + opt(&base.target), + opt(&candidate.target) + )); + } + if base.data_type != candidate.data_type { + changes.push(format!( + "type: {} -> {}", + opt(&base.data_type), + opt(&candidate.data_type) + )); + } + if base.nullability != candidate.nullability { + changes.push(format!( + "nullability: {} -> {}", + opt(&base.nullability), + opt(&candidate.nullability) + )); + } + if base.confidence != candidate.confidence { + changes.push(format!( + "confidence: {:?} -> {:?}", + base.confidence, candidate.confidence + )); + } + if base.generated != candidate.generated { + changes.push(format!( + "generated: {} -> {}", + base.generated, candidate.generated + )); + } + if base.materialization != candidate.materialization { + changes.push(format!( + "materialization: {} -> {}", + opt(&base.materialization), + opt(&candidate.materialization) + )); + } + if base.workflow != candidate.workflow { + changes.push(format!( + "workflow: {} -> {}", + opt(&base.workflow), + opt(&candidate.workflow) + )); + } + changes +} + +fn dataset_kind_label(kind: Option) -> &'static str { + match kind { + Some(crate::lineage::DatasetKind::Model) => "model", + Some(crate::lineage::DatasetKind::Source) => "source", + Some(crate::lineage::DatasetKind::Seed) => "seed", + Some(crate::lineage::DatasetKind::External) => "external", + None => "-", + } +} + +/// `field: old -> new` descriptions for every edge field that differs. +fn edge_changes(base: &LineageEdge, candidate: &LineageEdge) -> Vec { + let mut changes = Vec::new(); + if base.directness != candidate.directness { + changes.push(format!( + "directness: {:?} -> {:?}", + base.directness, candidate.directness + )); + } + if base.transformation != candidate.transformation { + changes.push(format!( + "transformation: {:?} -> {:?}", + base.transformation, candidate.transformation + )); + } + if base.confidence != candidate.confidence { + changes.push(format!( + "confidence: {:?} -> {:?}", + base.confidence, candidate.confidence + )); + } + if base.expression != candidate.expression { + changes.push(format!( + "expression: {} -> {}", + truncate(base.expression.as_deref().unwrap_or("-"), 60), + truncate(candidate.expression.as_deref().unwrap_or("-"), 60) + )); + } + changes +} + +/// `text` truncated to `max` characters — character-safe, never slicing +/// inside a UTF-8 sequence. +fn truncate(text: &str, max: usize) -> String { + if text.chars().count() <= max { + return text.to_string(); + } + format!("{}…", text.chars().take(max).collect::()) +} + +fn short(hash: &str) -> String { + hash.chars().take(12).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::compile::compile; + use crate::identity::ModelId; + use crate::model::{SemanticModel, SemanticProject}; + + fn graph(models: Vec) -> LineageGraph { + let compilation = compile(&SemanticProject::in_memory(models)); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + LineageGraph::build(&compilation) + } + + fn model(name: &str, sql: &str) -> SemanticModel { + SemanticModel::in_memory(ModelId::parse(name).expect("model id"), sql) + } + + /// names of all entries in a node list. + fn names(nodes: &[DiffNode]) -> Vec<&str> { + nodes.iter().map(|node| node.name.as_str()).collect() + } + + #[test] + fn identical_graphs_diff_empty() { + let base = graph(vec![model("a.one", "select 1 as id")]); + let candidate = graph(vec![model("a.one", "select 1 as id")]); + let diff = lineage_diff(&base, &candidate); + assert!(diff.is_empty(), "{diff:?}"); + } + + /// The fingerprint is the candidate identity promotion audits against: + /// identical definitions hash identically, and any change to the graph + /// — here a changed column expression — moves it. + #[test] + fn fingerprint_tracks_graph_identity() { + let one = graph(vec![ + model("a.up", "select 1 as id"), + model("a.down", "select id from a.up"), + ]); + let same = graph(vec![ + model("a.up", "select 1 as id"), + model("a.down", "select id from a.up"), + ]); + assert_eq!(one.fingerprint(), same.fingerprint()); + + let changed = graph(vec![ + model("a.up", "select 1 as id"), + model("a.down", "select id, id + 1 as next from a.up"), + ]); + assert_ne!(one.fingerprint(), changed.fingerprint()); + } + + #[test] + fn added_and_removed_models_report() { + let base = graph(vec![model("a.one", "select 1 as id")]); + let candidate = graph(vec![model("a.two", "select 1 as id")]); + let diff = lineage_diff(&base, &candidate); + assert!(names(&diff.nodes_added).contains(&"a.two")); + assert!(names(&diff.nodes_removed).contains(&"a.one")); + // The output datasets move with their models. + assert!(names(&diff.nodes_added).contains(&"a.two")); + assert!(names(&diff.nodes_removed).contains(&"a.one")); + } + + #[test] + fn changed_model_reports_version_change() { + let base = graph(vec![model("a.one", "select 1 as id")]); + let candidate = graph(vec![model("a.one", "select 2 as id")]); + let diff = lineage_diff(&base, &candidate); + assert_eq!(diff.nodes_changed.len(), 1); + assert_eq!(diff.nodes_changed[0].name, "a.one"); + assert!( + diff.nodes_changed[0] + .changes + .iter() + .any(|change| change.starts_with("version:")), + "{:?}", + diff.nodes_changed[0].changes + ); + } + + #[test] + fn new_dependency_edge_reports() { + let base = graph(vec![ + model("a.up", "select 1 as id"), + model("a.down", "select 1 as id"), + ]); + let candidate = graph(vec![ + model("a.up", "select 1 as id"), + model("a.down", "select * from a.up"), + ]); + let diff = lineage_diff(&base, &candidate); + // `a.down` now reads `a.up`'s output dataset — an `input` edge and + // the `derives` rollups appear on the candidate side only. + let added: Vec<(&str, &str)> = diff + .edges_added + .iter() + .map(|edge| (edge.kind.as_str(), edge.to.as_str())) + .collect(); + assert!( + added + .iter() + .any(|(kind, to)| *kind == "input" && to.contains("model://a/down")), + "{:?}", + diff.edges_added + ); + assert!(diff.edges_removed.is_empty()); + } + + #[test] + fn removed_model_orphans_its_consumers() { + // Base: down reads up. Candidate: up is gone — down remains and is + // orphaned. + let base = graph(vec![ + model("a.up", "select 1 as id"), + model("a.down", "select * from a.up"), + ]); + let candidate = graph(vec![model("a.down", "select * from a.up")]); + let diff = lineage_diff(&base, &candidate); + let impact = diff + .impacts + .iter() + .find(|impact| impact.node.contains("a/up")) + .expect("impact for a.up"); + assert!( + impact.orphans.iter().any(|name| name == "a.down"), + "{:?}", + impact + ); + } + + #[test] + fn column_lineage_change_reports() { + // Same model, different expression for the same output column — + // the derives edge's expression metadata changes. Exact + // assertions: a node `version:` change alone must not satisfy this. + let base = graph(vec![ + model("a.src", "select 1 as x"), + model("a.down", "select x + 1 as y from a.src"), + ]); + let candidate = graph(vec![ + model("a.src", "select 1 as x"), + model("a.down", "select x + 2 as y from a.src"), + ]); + let diff = lineage_diff(&base, &candidate); + let change = diff + .edges_changed + .iter() + .find(|edge| edge.kind == "derives" && edge.to.contains("a/down#y")) + .expect("changed derives edge into a.down.y"); + assert!( + change.from.contains("a/src#x"), + "expected the src.x -> down.y edge, got {change:?}" + ); + assert!( + change + .changes + .iter() + .any(|field| field.starts_with("expression:")), + "expected an expression change, got {:?}", + change.changes + ); + } + + #[test] + fn parallel_edges_same_endpoints_compare_as_a_bag() { + // `id` feeds `y` twice: directly through the SELECT and indirectly + // through the WHERE filter — two `derives` edges between the same + // column nodes, distinguished only by metadata. + let base = graph(vec![ + model("a.src", "select 1 as id"), + model("a.down", "select id as y from a.src where id > 0"), + ]); + let derives = |graph: &LineageGraph| { + graph + .edges() + .into_iter() + .filter(|(from, to, edge)| { + edge.kind == LineageEdgeKind::Derives + && from.uri().ends_with("a/src#id") + && to.uri().ends_with("a/down#y") + }) + .count() + }; + assert_eq!(derives(&base), 2, "base should carry parallel edges"); + + // Tighten only the filter: the direct edge is untouched, the + // indirect edge's expression moves. + let candidate = graph(vec![ + model("a.src", "select 1 as id"), + model("a.down", "select id as y from a.src where id > 5"), + ]); + assert_eq!(derives(&candidate), 2); + let diff = lineage_diff(&base, &candidate); + let changed: Vec<&EdgeChange> = diff + .edges_changed + .iter() + .filter(|edge| edge.kind == "derives" && edge.to.contains("a/down#y")) + .collect(); + assert_eq!( + changed.len(), + 1, + "exactly the filter edge should move, got {:?}", + diff.edges_changed + ); + assert!( + changed[0] + .changes + .iter() + .any(|field| field.starts_with("expression:")), + "{:?}", + changed[0].changes + ); + // The surviving direct edge is not reported removed or re-added. + assert!( + !diff + .edges_removed + .iter() + .chain(diff.edges_added.iter()) + .any(|edge| edge.to.contains("a/down#y")), + "{:?} / {:?}", + diff.edges_removed, + diff.edges_added + ); + } + + #[test] + fn removed_dependency_edge_reports_impact() { + // `a.down` stops reading `a.up`: every node survives, but the + // lineage path disappears — an edge removal with downstream impact. + let base = graph(vec![ + model("a.up", "select 1 as id"), + model("a.mid", "select id from a.up"), + model("a.down", "select id from a.mid"), + ]); + let candidate = graph(vec![ + model("a.up", "select 1 as id"), + model("a.mid", "select 1 as id"), + model("a.down", "select id from a.mid"), + ]); + let diff = lineage_diff(&base, &candidate); + let removed = diff + .edges_removed + .iter() + .find(|edge| { + edge.kind == "input" && edge.from.contains("a/up") && edge.to.contains("a/mid") + }) + .expect("a.up -> a.mid input edge removed"); + assert!(removed.detail.is_none(), "input edges carry no metadata"); + let impact = diff + .edge_impacts + .iter() + .find(|impact| { + impact.from.contains("a/up") + && impact.to.contains("a/mid") + && impact.change == "removed" + }) + .expect("edge impact for the removed a.up -> a.mid path"); + assert!( + impact.downstream.iter().any(|name| name == "a.down"), + "a.down loses the a.up contribution, got {:?}", + impact.downstream + ); + } + + #[test] + fn utf8_expressions_truncate_without_panicking() { + // A multibyte expression long enough to hit the 60-character + // shortening path — byte slicing would split a code point. + let padding = "é".repeat(40); + let sql = format!("select '{padding}' || x as y from a.src"); + let base = graph(vec![ + model("a.src", "select 1 as x"), + model("a.down", "select x as y from a.src"), + ]); + let candidate = graph(vec![model("a.src", "select 1 as x"), model("a.down", &sql)]); + let diff = lineage_diff(&base, &candidate); + assert!( + !diff.edges_changed.is_empty() || !diff.edges_added.is_empty(), + "expected the derives edge to move, got {diff:?}" + ); + } +} diff --git a/crates/phlo-transform-engine/src/artifacts.rs b/crates/phlo-transform-engine/src/artifacts.rs index b08f601..8760604 100644 --- a/crates/phlo-transform-engine/src/artifacts.rs +++ b/crates/phlo-transform-engine/src/artifacts.rs @@ -87,6 +87,75 @@ pub struct EnvironmentArtifact { pub environment: crate::environment::EnvironmentSetup, } +/// `lineage_diff.json` — the semantic graph diff between a base ref's +/// compiled lineage and the candidate side's, bound to the identities both +/// sides carried when the diff was produced so a stale report cannot be +/// silently treated as current. +#[derive(Clone, Debug, Serialize, serde::Deserialize)] +pub struct LineageDiffArtifact { + pub schema_version: u32, + /// How the base side was chosen: `merge-base` (the branch-change + /// workflow — `merge-base(ref, HEAD)` vs the workspace) or `ref` (an + /// exact ref→ref comparison). + #[serde(default = "default_base_kind_ref")] + pub base_kind: String, + /// The git ref the diff was requested against, as given. + pub base_ref: String, + /// The commit the base graph was actually compiled from — the + /// merge-base for a workspace diff, the ref's head for ref→ref. + pub base_commit: String, + /// Candidate-side identity: the git head and worktree state the diff + /// was produced against. + #[serde(default)] + pub candidate: CandidateProvenance, + /// The Nessie branch pair this diff was produced for, when both refs + /// resolved at diff time. Absent means the artifact was never bound to + /// an environment — promotion can only treat it as advisory. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub environment: Option, + pub diff: phlo_transform_core::LineageDiff, +} + +fn default_base_kind_ref() -> String { + // Artifacts written before provenance existed resolved the ref + // directly — `ref`, not `merge-base`. + "ref".to_string() +} + +/// The candidate side's Git identity when a lineage diff was produced. +#[derive(Clone, Debug, Default, Serialize, serde::Deserialize)] +pub struct CandidateProvenance { + /// The name identifying the candidate (`--ref`/`--environment`), when + /// one was given — `None` for an anonymous worktree. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub git_ref: Option, + /// HEAD's commit at diff time (worktree diffs), or the candidate ref's + /// resolved commit (ref→ref diffs). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub head: Option, + /// Whether the diffed candidate included uncommitted worktree changes. + #[serde(default)] + pub dirty: bool, + /// Fingerprint of the candidate's canonical lineage graph — binds the + /// artifact to the exact definitions it was produced from, so code + /// edited after the diff cannot pass as audited. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lineage_hash: Option, + /// Model name → content-addressed version hash on the candidate side — + /// diagnostic context for the diff's definitional identity. + #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] + pub model_versions: std::collections::BTreeMap, +} + +/// The Nessie branch pair a lineage diff was produced for. +#[derive(Clone, Debug, Serialize, serde::Deserialize)] +pub struct LineageEnvironment { + pub candidate_ref: String, + pub candidate_hash: String, + pub target_ref: String, + pub target_hash: String, +} + /// `lineage.json` — the canonical lineage graph document. Every model, /// dataset, column and test node, every edge, and the directness / /// transformation / confidence metadata on column edges. @@ -247,6 +316,11 @@ impl ArtifactWriter { ) } + /// Write `lineage_diff.json`. + pub fn write_lineage_diff(&self, artifact: &LineageDiffArtifact) -> Result<(), EngineError> { + self.write("lineage_diff", artifact) + } + fn write(&self, name: &str, value: &T) -> Result<(), EngineError> { std::fs::create_dir_all(&self.directory) .map_err(|error| EngineError::Artifact(error.to_string()))?; diff --git a/crates/phlo-transform-engine/src/lib.rs b/crates/phlo-transform-engine/src/lib.rs index 2b9fccf..a71e76b 100644 --- a/crates/phlo-transform-engine/src/lib.rs +++ b/crates/phlo-transform-engine/src/lib.rs @@ -28,9 +28,9 @@ pub mod util; pub use adapter::{Adapter, CatalogRequest, ColumnInfo, QueryResult}; pub use artifacts::{ - ArtifactWriter, DiffArtifact, EnvironmentArtifact, GraphArtifactFile, LineageArtifact, - ManifestArtifact, OpenLineageArtifact, PlanArtifact, PromotionArtifact, RunArtifact, - SCHEMA_VERSION, + ArtifactWriter, CandidateProvenance, DiffArtifact, EnvironmentArtifact, GraphArtifactFile, + LineageArtifact, LineageDiffArtifact, LineageEnvironment, ManifestArtifact, + OpenLineageArtifact, PlanArtifact, PromotionArtifact, RunArtifact, SCHEMA_VERSION, }; pub use branch_diff::{ branch_diff, materialized_for_environment, model_keys, retarget, seeds_for_environment, diff --git a/docs/engine.md b/docs/engine.md index 4838836..24c9856 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -361,6 +361,7 @@ Written under `.phlo/transform/` with `schema_version = 3`: | `run.json` | run id, plan id, `continued_from`, status, per-status counts, model/seed/test results with structured failures and attempts, events | | `environment.json` | provisioned base/candidate references and candidate catalog | | `diff.json` | data/schema diff report (strategy, rows, columns, partitions, policies) | +| `lineage_diff.json` | semantic lineage diff vs a Git baseline (nodes/edges added, removed, changed; orphaned consumers and affected downstream paths) bound to `base_kind`/`base_commit`, the candidate's git head + worktree state, and the Nessie pair when resolved | | `promotion.json` | promotion id, references/hashes, gates, conflicts, timestamp | `schema_version` makes the interface explicit and versionable. diff --git a/docs/lineage.md b/docs/lineage.md index b3972b0..25fb4a0 100644 --- a/docs/lineage.md +++ b/docs/lineage.md @@ -141,6 +141,8 @@ phlo-transform lineage assay.results.titre # column: direct/indirect/transi phlo-transform lineage --format graph # canonical document (JSON) phlo-transform lineage assay.results --format graph # scoped document phlo-transform lineage --format openlineage # OpenLineage export +phlo-transform lineage --diff main # semantic diff vs merge-base(main, HEAD) +phlo-transform lineage --diff main feature/foo # exact ref → ref comparison phlo-transform impact assay.results # downstream models/tests phlo-transform impact assay.results.titre # downstream columns/models/tests phlo-transform impact external.samples.volume # source-column impact @@ -152,6 +154,77 @@ Human column output prints `Confidence:` when it is not `exact` and lists regardless of `--json`; selector terms scope it the same way they scope `plan`/`run`. +## Diffing lineage across refs + +`lineage --diff ` answers "what does this branch change about the +graph itself" — the semantic complement to `branch_diff`'s data comparison. +One ref uses the same merge-base semantics as `--since`: the base is +`merge-base(ref, HEAD)`, so a feature branch is compared against where it +diverged — never against the other ref's current head. Two refs +(`lineage --diff main feature/foo`) compare the exact refs with no worktree +involved. The base side is compiled for real: the workspace subtree at the +baseline commit is materialised into a temporary directory (`git ls-tree` + +`cat-file`, never a worktree, so the checkout is untouched), discovered and +compiled with the same options, then the two `LineageGraph`s are compared +node-for-node and edge-for-edge. Edges are a multiset — parallel edges +between the same two nodes (say, a column that is both selected and +filtered on) are compared by their full metadata, not collapsed: + +- `nodes_added` / `nodes_removed` — models, datasets, columns and tests + that exist on only one side; +- `nodes_changed` — the field-level delta (`version`, `materialization`, + `target`, `type`, `path`, …) for nodes present on both sides; +- `edges_added` / `edges_removed` — dependency structure: a model that + started (or stopped) reading a dataset is a changed `input` edge, not + just a changed file; +- `edges_changed` — column-level `derives` edges whose `transformation`, + `directness`, `confidence` or `expression` moved; +- `impacts` — for every removed node, the base-side consumers it orphans + (the models and tests that were reading it); +- `edge_impacts` — for every removed or metadata-moved edge, the base-side + downstream that loses or alters that lineage path, so dropping a + dependency while both nodes survive still reports impact. + +```text +Lineage diff vs main (merge-base 1a2b3c4d5e6f) + +Removed + - assay.legacy (model) + - assay.legacy (dataset) + +Changed + ~ assay.results (model) + version: 3f2a1c9d… -> 9be21f7a… + +Edges + + dataset://assay/clean --input-> model://assay/results + - dataset://assay/legacy --input-> model://assay/results + +Impacts + model://assay/legacy orphans: assay.report, test://legacy_range +``` + +The report is written to `.phlo/transform/lineage_diff.json`; `--json` +prints the same document. Beyond the diff itself the artifact binds its +provenance: `base_kind` (`merge-base` or `ref`), the resolved `base_commit`, +the candidate's git head and worktree state, a `lineage_hash` fingerprint +of the candidate's canonical graph, and — when `--ref`/`--from` +names a Nessie environment that resolves — the candidate and target branch +hashes it was produced for. `promote` reads that binding: a lineage artifact +covering this candidate and target at their current commits *and* whose +fingerprint still matches the compiled workspace reports +`current`; one produced for another pair, before either side moved, before +the candidate's definitions changed, or naming refs that no longer resolve, +reports `stale` with the reason rather than standing as evidence; an +unbound artifact is advisory only. A base ref that fails to load or compile +reports its diagnostics and exits non-zero rather than diffing a partial +graph. + +In the branch workflow this is the code-side review step: `branch_diff` +proves what the data would look like after promotion, `lineage --diff` +proves what the *definitions* would look like — new dependencies, dropped +models, orphaned consumers — before `promote` merges the data branch. + ## OpenLineage export `phlo-transform-openlineage` maps the graph to a design-time document — a diff --git a/docs/wap.md b/docs/wap.md index b1b5635..fa5e618 100644 --- a/docs/wap.md +++ b/docs/wap.md @@ -81,6 +81,19 @@ A failed run or audit leaves the candidate isolated and does not advance diff, gate evaluation, promotion, stale-promotion rejection and branch cleanup against Nessie + Trino/Iceberg. +Two reviews precede promotion, one per axis: + +- `lineage --diff ` — the **code-side** delta: models, datasets, + columns, tests and dependency edges added, removed or changed relative + to the merge-base the work branched from, plus the consumers each + removal orphans and the downstream each lost lineage path affects. + Persisted to `lineage_diff.json` and bound to the candidate; `promote` + surfaces it as `current`, `advisory` or `stale`. See `docs/lineage.md`. +- `diff --from --to ` — the **data-side** delta: + dataset classification, schema and contract changes, row counts and + keyed value diffs between the two Nessie references. Persisted to + `branch_diff.json`. See below. + ## Branch diff ```bash