Skip to content

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

  1. Start with a v1 spec (basic email matching)
  2. Evolve to v2 (add fuzzy name matching, adjust thresholds)
  3. Evolve to v3 (add phone rule, change survivorship strategy)
  4. Use diff() to track changes between each version
  5. Use plan hashes for change detection in automation
  6. Set up a pre-commit hook for spec validation
  7. 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:

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.60

Validate and capture the plan hash:

python
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: 56

Step 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:

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.50

Changes from v1:

  • Added name_fuzzy rule (Jaro-Winkler, weight 0.6)
  • Lowered review threshold 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:

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.45

Changes from v2:

  • Added support source (Zendesk)
  • Added phone_fuzzy rule (Levenshtein, weight 0.5)
  • Changed survivorship default from source_priority to most_complete (removed overrides)
  • Lowered match threshold from 0.85 to 0.80
  • Lowered review threshold 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

python
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.5

Diff v2 to v3

python
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_complete

Full Diff: v1 to v3

You can also diff across multiple versions to see the total cumulative change:

python
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_complete

Step 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.

python
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:

python
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_complete

Step 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):

bash
#!/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:

bash
chmod +x .git/hooks/pre-commit

Option B: Pre-Commit Framework

If you use the pre-commit framework, add to .pre-commit-config.yaml:

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

bash
# 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:

yaml
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}")
          PYTHON

What This Pipeline Does

  1. Triggers on any PR that modifies spec YAML files.
  2. Validates every spec file in the repository. A single invalid spec fails the build.
  3. 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_complete

Step 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.

python
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

StepActionCommand
1Edit spec YAML(text editor)
2Validate locallyvalidate(spec).raise_on_error()
3Preview planplan(spec)
4Diff against previousdiff(old_spec, new_spec)
5CommitPre-commit hook validates automatically
6Open PRCI validates and prints diff summary
7MergePlan hash detects change, triggers reconciliation
8RunReconcile 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

python
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

The identity and delegation layer for AI agents.