Tutorial: Spec Evolution
Specs evolve over time as you refine matching logic, add new data sources, and adjust thresholds. This tutorial walks through the lifecycle of a spec across three versions, shows how to diff changes between them, and demonstrates how to integrate spec validation into a CI/CD pipeline.
Time: 10 minutes Prerequisites: Python 3.9+, pip install kanoniv
What You Will Build
- Start with a v1 spec (basic email matching)
- Evolve to v2 (add fuzzy name matching, adjust thresholds)
- Evolve to v3 (add phone rule, change survivorship strategy)
- Use
diff()to track changes between each version - Use plan hashes for change detection in automation
- Set up a pre-commit hook for spec validation
- Add a CI pipeline for automated validation
Step 1: Version 1 — Baseline Spec
Start with the simplest possible spec: exact email matching only.
Create spec-v1.yaml:
api_version: "kanoniv/v2"
identity_version: "customer_v1"
metadata:
name: customer-matching
description: "v1: Baseline email-only matching"
entity:
name: customer
sources:
- name: crm
adapter: csv
location: crm_contacts.csv
primary_key: id
attributes:
name: name
email: email
phone: phone
company: company
- name: billing
adapter: csv
location: stripe_customers.csv
primary_key: customer_id
attributes:
name: full_name
email: email
rules:
- name: email_exact
type: exact
field: email
weight: 1.0
survivorship:
default: source_priority
overrides:
- field: name
strategy: source_priority
priority: [crm, billing]
- field: email
strategy: source_priority
priority: [crm, billing]
decision:
thresholds:
match: 0.85
review: 0.60Validate and capture the plan hash:
from kanoniv import Spec, validate, plan
spec_v1 = Spec.from_file("spec-v1.yaml")
validate(spec_v1).raise_on_error()
plan_v1 = plan(spec_v1)
print(f"v1 plan hash: {plan_v1.hash}")
print(f"v1 rules: {plan_v1.rule_count}")
print(f"v1 estimated comparisons: {plan_v1.estimated_comparisons}")Expected output:
v1 plan hash: a1b2c3d4
v1 rules: 1
v1 estimated comparisons: 56Step 2: Version 2 — Add Fuzzy Name Matching
After reviewing v1 results, you discover that some records have different email addresses but clearly refer to the same person (e.g., "John Doe" in the CRM and "Jonathan Doe" in billing). Add a fuzzy name rule and adjust thresholds.
Create spec-v2.yaml:
api_version: "kanoniv/v2"
identity_version: "customer_v2"
metadata:
name: customer-matching
description: "v2: Add fuzzy name matching, lower review threshold"
entity:
name: customer
sources:
- name: crm
adapter: csv
location: crm_contacts.csv
primary_key: id
attributes:
name: name
email: email
phone: phone
company: company
- name: billing
adapter: csv
location: stripe_customers.csv
primary_key: customer_id
attributes:
name: full_name
email: email
rules:
- name: email_exact
type: exact
field: email
weight: 1.0
- name: name_fuzzy
type: similarity
field: name
algorithm: jaro_winkler
threshold: 0.82
weight: 0.6
survivorship:
default: source_priority
overrides:
- field: name
strategy: source_priority
priority: [crm, billing]
- field: email
strategy: source_priority
priority: [crm, billing]
decision:
thresholds:
match: 0.85
review: 0.50Changes from v1:
- Added
name_fuzzyrule (Jaro-Winkler, weight 0.6) - Lowered
reviewthreshold from 0.60 to 0.50
Step 3: Version 3 — Add Phone Rule, Change Survivorship
Further analysis shows that some records share phone numbers but have different emails and name variations. Add a phone rule and switch to most_complete survivorship since the billing source sometimes has more complete data.
Create spec-v3.yaml:
api_version: "kanoniv/v2"
identity_version: "customer_v3"
metadata:
name: customer-matching
description: "v3: Add phone matching, switch to most_complete survivorship"
entity:
name: customer
sources:
- name: crm
adapter: csv
location: crm_contacts.csv
primary_key: id
attributes:
name: name
email: email
phone: phone
company: company
- name: billing
adapter: csv
location: stripe_customers.csv
primary_key: customer_id
attributes:
name: full_name
email: email
- name: support
adapter: csv
location: zendesk_users.csv
primary_key: user_id
attributes:
name: display_name
email: email_address
company: organization
rules:
- name: email_exact
type: exact
field: email
weight: 1.0
- name: name_fuzzy
type: similarity
field: name
algorithm: jaro_winkler
threshold: 0.82
weight: 0.6
- name: phone_fuzzy
type: similarity
field: phone
algorithm: levenshtein
threshold: 0.85
weight: 0.5
survivorship:
default: most_complete
decision:
thresholds:
match: 0.80
review: 0.45Changes from v2:
- Added
supportsource (Zendesk) - Added
phone_fuzzyrule (Levenshtein, weight 0.5) - Changed survivorship
defaultfromsource_prioritytomost_complete(removed overrides) - Lowered
matchthreshold from 0.85 to 0.80 - Lowered
reviewthreshold from 0.50 to 0.45
Step 4: Diff Between Versions
The diff() function compares two specs and returns a structured summary of what changed.
Diff v1 to v2
from kanoniv import Spec, diff
spec_v1 = Spec.from_file("spec-v1.yaml")
spec_v2 = Spec.from_file("spec-v2.yaml")
diff_1_to_2 = diff(spec_v1, spec_v2)
print("=== Diff: v1 -> v2 ===\n")
print(f"Has changes: {diff_1_to_2.has_changes}")
print()
if diff_1_to_2.sources_added:
print(f"Sources added: {diff_1_to_2.sources_added}")
if diff_1_to_2.sources_removed:
print(f"Sources removed: {diff_1_to_2.sources_removed}")
if diff_1_to_2.rules_added:
print(f"Rules added: {diff_1_to_2.rules_added}")
if diff_1_to_2.rules_removed:
print(f"Rules removed: {diff_1_to_2.rules_removed}")
if diff_1_to_2.rules_modified:
print(f"Rules modified: {diff_1_to_2.rules_modified}")
if diff_1_to_2.thresholds_changed:
print(f"Thresholds changed:")
for name, change in diff_1_to_2.thresholds_changed.items():
print(f" {name}: {change.old} -> {change.new}")
if diff_1_to_2.survivorship_changed:
print(f"Survivorship changed: {diff_1_to_2.survivorship_changed}")Expected output:
=== Diff: v1 -> v2 ===
Has changes: True
Rules added: ['name_fuzzy']
Thresholds changed:
review: 0.6 -> 0.5Diff v2 to v3
spec_v3 = Spec.from_file("spec-v3.yaml")
diff_2_to_3 = diff(spec_v2, spec_v3)
print("=== Diff: v2 -> v3 ===\n")
print(f"Has changes: {diff_2_to_3.has_changes}")
print()
if diff_2_to_3.sources_added:
print(f"Sources added: {diff_2_to_3.sources_added}")
if diff_2_to_3.sources_removed:
print(f"Sources removed: {diff_2_to_3.sources_removed}")
if diff_2_to_3.rules_added:
print(f"Rules added: {diff_2_to_3.rules_added}")
if diff_2_to_3.rules_removed:
print(f"Rules removed: {diff_2_to_3.rules_removed}")
if diff_2_to_3.rules_modified:
print(f"Rules modified: {diff_2_to_3.rules_modified}")
if diff_2_to_3.thresholds_changed:
print(f"Thresholds changed:")
for name, change in diff_2_to_3.thresholds_changed.items():
print(f" {name}: {change.old} -> {change.new}")
if diff_2_to_3.survivorship_changed:
print(f"Survivorship changed: {diff_2_to_3.survivorship_changed}")Expected output:
=== Diff: v2 -> v3 ===
Has changes: True
Sources added: ['support']
Rules added: ['phone_fuzzy']
Thresholds changed:
match: 0.85 -> 0.8
review: 0.5 -> 0.45
Survivorship changed: source_priority -> most_completeFull Diff: v1 to v3
You can also diff across multiple versions to see the total cumulative change:
diff_1_to_3 = diff(spec_v1, spec_v3)
print("=== Diff: v1 -> v3 (cumulative) ===\n")
print(f"Sources added: {diff_1_to_3.sources_added}")
print(f"Rules added: {diff_1_to_3.rules_added}")
print(f"Thresholds changed:")
for name, change in diff_1_to_3.thresholds_changed.items():
print(f" {name}: {change.old} -> {change.new}")
print(f"Survivorship changed: {diff_1_to_3.survivorship_changed}")Expected output:
=== Diff: v1 -> v3 (cumulative) ===
Sources added: ['support']
Rules added: ['name_fuzzy', 'phone_fuzzy']
Thresholds changed:
match: 0.85 -> 0.8
review: 0.6 -> 0.45
Survivorship changed: source_priority -> most_completeStep 5: Plan Hashes for Change Detection
Every spec produces a deterministic plan hash. When the spec changes in a way that affects reconciliation behavior, the hash changes. When cosmetic changes are made (description, comments), the hash stays the same.
from kanoniv import plan
plan_v1 = plan(spec_v1)
plan_v2 = plan(spec_v2)
plan_v3 = plan(spec_v3)
print(f"v1 hash: {plan_v1.hash}")
print(f"v2 hash: {plan_v2.hash}")
print(f"v3 hash: {plan_v3.hash}")
# All different because rules/thresholds changed
assert plan_v1.hash != plan_v2.hash
assert plan_v2.hash != plan_v3.hash
# Cosmetic change does not affect hash
spec_v3_copy = Spec.from_file("spec-v3.yaml")
spec_v3_copy.description = "Updated description, same logic"
plan_v3_copy = plan(spec_v3_copy)
print(f"\nv3 hash (original): {plan_v3.hash}")
print(f"v3 hash (new description): {plan_v3_copy.hash}")
assert plan_v3.hash == plan_v3_copy.hash
print("Hashes match — description change is cosmetic.")Expected output:
v1 hash: a1b2c3d4
v2 hash: e5f6a7b8
v3 hash: c9d0e1f2
v3 hash (original): c9d0e1f2
v3 hash (new description): c9d0e1f2
Hashes match — description change is cosmetic.Plan Hashes in Automation
Store the plan hash alongside reconciliation results. Before re-running, compare hashes — if the hash has not changed, the spec logic is identical and you can skip re-processing. This is especially valuable for large datasets where reconciliation is expensive.
Step 6: Version Tracking Script
Here is a complete script that validates all versions, diffs between them, and summarizes the evolution:
from kanoniv import Spec, validate, plan, diff
versions = [
("v1", "spec-v1.yaml"),
("v2", "spec-v2.yaml"),
("v3", "spec-v3.yaml"),
]
# Load and validate all
specs = {}
plans = {}
for label, path in versions:
spec = Spec.from_file(path)
validate(spec).raise_on_error()
p = plan(spec)
specs[label] = spec
plans[label] = p
print(f"{label}: hash={p.hash} rules={p.rule_count} "
f"sources={p.source_count} comparisons={p.estimated_comparisons}")
print()
# Diff consecutive versions
for i in range(len(versions) - 1):
label_a = versions[i][0]
label_b = versions[i + 1][0]
d = diff(specs[label_a], specs[label_b])
print(f"--- {label_a} -> {label_b} ---")
if d.sources_added:
print(f" + sources: {', '.join(d.sources_added)}")
if d.sources_removed:
print(f" - sources: {', '.join(d.sources_removed)}")
if d.rules_added:
print(f" + rules: {', '.join(d.rules_added)}")
if d.rules_removed:
print(f" - rules: {', '.join(d.rules_removed)}")
if d.rules_modified:
print(f" ~ rules: {', '.join(d.rules_modified)}")
if d.thresholds_changed:
for name, change in d.thresholds_changed.items():
print(f" ~ {name}: {change.old} -> {change.new}")
if d.survivorship_changed:
print(f" ~ survivorship: {d.survivorship_changed}")
print()Expected output:
v1: hash=a1b2c3d4 rules=1 sources=2 comparisons=56
v2: hash=e5f6a7b8 rules=2 sources=2 comparisons=56
v3: hash=c9d0e1f2 rules=3 sources=3 comparisons=176
--- v1 -> v2 ---
+ rules: name_fuzzy
~ review: 0.6 -> 0.5
--- v2 -> v3 ---
+ sources: support
+ rules: phone_fuzzy
~ match: 0.85 -> 0.8
~ review: 0.5 -> 0.45
~ survivorship: source_priority -> most_completeStep 7: Pre-Commit Hook
Add a pre-commit hook that validates all spec files before they can be committed. This catches errors before they reach CI.
Option A: Shell Script Hook
Create .git/hooks/pre-commit (or add to your existing hook):
#!/bin/bash
set -e
# Find all YAML spec files staged for commit
SPEC_FILES=$(git diff --cached --name-only --diff-filter=ACM | grep -E '\.ya?ml$' | grep -i spec || true)
if [ -z "$SPEC_FILES" ]; then
exit 0
fi
echo "Validating Kanoniv specs..."
for file in $SPEC_FILES; do
echo " Checking $file..."
python -c "
from kanoniv import Spec, validate
spec = Spec.from_file('$file')
result = validate(spec)
result.raise_on_error()
print(' Valid.')
"
done
echo "All specs valid."Make it executable:
chmod +x .git/hooks/pre-commitOption B: Pre-Commit Framework
If you use the pre-commit framework, add to .pre-commit-config.yaml:
repos:
- repo: local
hooks:
- id: validate-kanoniv-specs
name: Validate Kanoniv specs
entry: python -c "
import sys, glob;
from kanoniv import Spec, validate;
files = sys.argv[1:];
[validate(Spec.from_file(f)).raise_on_error() for f in files];
print(f'Validated {len(files)} spec(s).')
"
language: python
files: '.*spec.*\.ya?ml$'
additional_dependencies: ['kanoniv']Testing the Hook
# Make a spec with an error
cat > bad-spec.yaml << 'EOF'
api_version: "kanoniv/v2"
identity_version: "broken_v1"
metadata:
name: broken
entity:
name: customer
sources:
- name: crm
adapter: csv
location: data.csv
primary_key: id
rules:
- name: bad_rule
type: exact
field: nonexistent_field
weight: 1.0
decision:
thresholds:
match: 0.85
review: 0.60
EOF
git add bad-spec.yaml
git commit -m "This will fail"Expected output:
Validating Kanoniv specs...
Checking bad-spec.yaml...
SpecValidationError: Rule 'bad_rule' references field 'nonexistent_field'
which is not defined in any source.
Commit aborted.Step 8: CI Pipeline with GitHub Actions
Add automated spec validation and diff reporting to your CI/CD pipeline.
Create .github/workflows/spec-validate.yml:
name: Validate Kanoniv Specs
on:
pull_request:
paths:
- '**/*spec*.yaml'
- '**/*spec*.yml'
jobs:
validate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
with:
fetch-depth: 0 # Need full history for diff
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Kanoniv
run: pip install kanoniv
- name: Validate all specs
run: |
python << 'PYTHON'
import glob
import sys
from kanoniv import Spec, validate, plan
specs = glob.glob("**/*spec*.yaml", recursive=True) + \
glob.glob("**/*spec*.yml", recursive=True)
if not specs:
print("No spec files found.")
sys.exit(0)
errors = []
for path in specs:
try:
spec = Spec.from_file(path)
validate(spec).raise_on_error()
p = plan(spec)
print(f"PASS: {path} (hash: {p.hash}, "
f"{p.rule_count} rules, "
f"{p.source_count} sources)")
except Exception as e:
errors.append((path, str(e)))
print(f"FAIL: {path} - {e}")
if errors:
print(f"\n{len(errors)} spec(s) failed validation.")
sys.exit(1)
else:
print(f"\nAll {len(specs)} spec(s) valid.")
PYTHON
- name: Diff against base branch
if: github.event_name == 'pull_request'
run: |
python << 'PYTHON'
import glob
import subprocess
from kanoniv import Spec, plan, diff
# Find spec files changed in this PR
result = subprocess.run(
["git", "diff", "--name-only", "origin/${{ github.base_ref }}", "HEAD"],
capture_output=True, text=True
)
changed = [f for f in result.stdout.strip().split("\n")
if f and ("spec" in f.lower()) and f.endswith((".yaml", ".yml"))]
if not changed:
print("No spec files changed in this PR.")
exit(0)
for path in changed:
print(f"\n=== {path} ===")
try:
# Load current version
current = Spec.from_file(path)
current_plan = plan(current)
# Load base version
base_content = subprocess.run(
["git", "show", f"origin/${{ github.base_ref }}:{path}"],
capture_output=True, text=True
)
if base_content.returncode != 0:
print(f" New spec file (no base version)")
print(f" Hash: {current_plan.hash}")
print(f" Rules: {current_plan.rule_count}")
print(f" Sources: {current_plan.source_count}")
continue
import tempfile, os
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml',
delete=False) as tmp:
tmp.write(base_content.stdout)
tmp_path = tmp.name
base = Spec.from_file(tmp_path)
os.unlink(tmp_path)
base_plan = plan(base)
changes = diff(base, current)
if not changes.has_changes:
print(f" No logical changes (hash: {current_plan.hash})")
continue
print(f" Hash: {base_plan.hash} -> {current_plan.hash}")
if changes.sources_added:
print(f" + Sources: {', '.join(changes.sources_added)}")
if changes.sources_removed:
print(f" - Sources: {', '.join(changes.sources_removed)}")
if changes.rules_added:
print(f" + Rules: {', '.join(changes.rules_added)}")
if changes.rules_removed:
print(f" - Rules: {', '.join(changes.rules_removed)}")
if changes.rules_modified:
print(f" ~ Rules: {', '.join(changes.rules_modified)}")
if changes.thresholds_changed:
for name, change in changes.thresholds_changed.items():
print(f" ~ {name}: {change.old} -> {change.new}")
if changes.survivorship_changed:
print(f" ~ Survivorship: {changes.survivorship_changed}")
except Exception as e:
print(f" Error: {e}")
PYTHONWhat This Pipeline Does
- Triggers on any PR that modifies spec YAML files.
- Validates every spec file in the repository. A single invalid spec fails the build.
- Diffs changed specs against the base branch and prints a human-readable summary of what changed — sources added, rules modified, thresholds adjusted.
Example CI output for a PR that updates a spec:
PASS: specs/customer-360.yaml (hash: c9d0e1f2, 3 rules, 3 sources)
PASS: specs/payment-dedup.yaml (hash: b8e4d1f3, 3 rules, 3 sources)
All 2 spec(s) valid.
=== specs/customer-360.yaml ===
Hash: e5f6a7b8 -> c9d0e1f2
+ Sources: support
+ Rules: phone_fuzzy
~ match: 0.85 -> 0.8
~ review: 0.5 -> 0.45
~ Survivorship: source_priority -> most_completeStep 9: Store Plan Hashes for Caching
In a production system, you can skip reconciliation if the spec has not changed since the last run. Store the plan hash alongside your results.
import json
import os
from kanoniv import Spec, Source, reconcile, plan, utcnow
CACHE_FILE = ".kanoniv-cache.json"
def load_cache():
if os.path.exists(CACHE_FILE):
with open(CACHE_FILE) as f:
return json.load(f)
return {}
def save_cache(cache):
with open(CACHE_FILE, "w") as f:
json.dump(cache, f, indent=2)
# Load spec and compute plan hash
spec = Spec.from_file("spec-v3.yaml")
execution = plan(spec)
cache = load_cache()
last_hash = cache.get("last_plan_hash")
if last_hash == execution.hash:
print(f"Spec unchanged (hash: {execution.hash}). Skipping reconciliation.")
print("Use --force to re-run.")
else:
print(f"Spec changed: {last_hash or '(first run)'} -> {execution.hash}")
print("Running reconciliation...")
sources = [
Source.from_csv("crm", "crm_contacts.csv"),
Source.from_csv("billing", "stripe_customers.csv"),
Source.from_csv("support", "zendesk_users.csv"),
]
result = reconcile(sources, spec)
print(f"Done. {len(result.golden_records)} golden records, "
f"merge rate {result.merge_rate:.1%}")
# Save hash
cache["last_plan_hash"] = execution.hash
cache["last_run"] = str(utcnow())
cache["golden_record_count"] = len(result.golden_records)
cache["merge_rate"] = result.merge_rate
save_cache(cache)
print(f"Cache updated: {CACHE_FILE}")Summary: Spec Evolution Workflow
| Step | Action | Command |
|---|---|---|
| 1 | Edit spec YAML | (text editor) |
| 2 | Validate locally | validate(spec).raise_on_error() |
| 3 | Preview plan | plan(spec) |
| 4 | Diff against previous | diff(old_spec, new_spec) |
| 5 | Commit | Pre-commit hook validates automatically |
| 6 | Open PR | CI validates and prints diff summary |
| 7 | Merge | Plan hash detects change, triggers reconciliation |
| 8 | Run | Reconcile only if plan hash changed |
This workflow ensures that spec changes are always validated, reviewed, and traceable — giving your team confidence that identity resolution logic evolves safely.
Full Working Script
from kanoniv import Spec, validate, plan, diff
# Load all versions
specs = {
"v1": Spec.from_file("spec-v1.yaml"),
"v2": Spec.from_file("spec-v2.yaml"),
"v3": Spec.from_file("spec-v3.yaml"),
}
# Validate all
for label, spec in specs.items():
validate(spec).raise_on_error()
p = plan(spec)
print(f"{label}: hash={p.hash} rules={p.rule_count} "
f"sources={p.source_count}")
# Diff chain
print("\n--- Evolution ---")
labels = list(specs.keys())
for i in range(len(labels) - 1):
a, b = labels[i], labels[i + 1]
d = diff(specs[a], specs[b])
print(f"\n{a} -> {b}:")
if d.sources_added:
print(f" + sources: {', '.join(d.sources_added)}")
if d.rules_added:
print(f" + rules: {', '.join(d.rules_added)}")
if d.thresholds_changed:
for name, change in d.thresholds_changed.items():
print(f" ~ {name}: {change.old} -> {change.new}")
if d.survivorship_changed:
print(f" ~ survivorship: {d.survivorship_changed}")
# Cumulative diff
print(f"\n--- Cumulative: {labels[0]} -> {labels[-1]} ---")
total_diff = diff(specs[labels[0]], specs[labels[-1]])
if total_diff.sources_added:
print(f" + sources: {', '.join(total_diff.sources_added)}")
if total_diff.rules_added:
print(f" + rules: {', '.join(total_diff.rules_added)}")
if total_diff.thresholds_changed:
for name, change in total_diff.thresholds_changed.items():
print(f" ~ {name}: {change.old} -> {change.new}")
if total_diff.survivorship_changed:
print(f" ~ survivorship: {total_diff.survivorship_changed}")Next Steps
- Customer 360 — Three-source matching with source-priority survivorship
- Payment Deduplication — Range-based matching for amounts and dates
- Lead Matching — Review queues and most-complete survivorship
- Spec Reference — Full documentation for all spec fields
- Versioning Overview — The four versioning layers: specs, rules, entities, and temporal
