phase 1 review fixes: dedup edges, real export, validation, tests

Review of aa2ca0a found and fixed: - re-ingest duplicated ALL edges (upsert conflicted on valid_from=now(), never fired) — migration 007 dedupes + partial unique index on current edges; upsert now targets it. Regression-tested. - export was a stub that overwrote seeds/*.yaml with 11-byte "version: 1" files — implemented real deterministic export (ontology/inventory/policy, cognition-layer excluded); round-trip is byte-stable (tested) - DB password leaked in startup logs (slog JSON bypasses String()) — Config now implements slog.LogValuer; regression-tested - docker-compose had literal '***' as DB password — env-interpolated - uuid.New() (v4) → uuid.NewV7() per ADR-0005 - no ontology validation on ingest — internal/ontology TypeTree: abstract instantiation rejected, relationship endpoints hierarchy-validated, cardinality enforced in-transaction, lifecycle states checked, default state applied (Phase 1 gate items, R3-1) - getOrCreateEntityID swallowed non-ErrNoRows errors - migration runner now holds a session advisory lock on one connection - Makefile: hardcoded /opt/homebrew/bin/go → go; test-db target Tests: 4 unit suites + 7 integration tests (env-guarded, throwaway DB per run): migrate idempotent, seed idempotent + no dup edges, abstract/edge/ cardinality rejection, blast_radius cycle termination, export round-trip. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-07 08:11:01 +02:00
parent aa2ca0ae6f
commit 1b04683639
13 changed files with 1250 additions and 70 deletions
--- a/internal/db/export.go
+++ b/internal/db/export.go
@@ -0,0 +1,300 @@
+package db
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"gopkg.in/yaml.v3"
+)
+
+// ExportToYAML regenerates the three seed YAMLs from the DB (DR / version
+// control, plan D6). Output is deterministic: maps marshal with sorted keys
+// (yaml.v3 default for map[string]any) and lists are ordered by slug/name,
+// so export → ingest → export is byte-stable.
+//
+// Cognition-layer entities (signals, executions, patterns, …) are runtime
+// state, not inventory — they are excluded from the export.
+func ExportToYAML(ctx context.Context, pool *Pool) (map[string][]byte, error) {
+	result := make(map[string][]byte)
+
+	for name, fn := range map[string]func(context.Context, *Pool) (map[string]any, error){
+		"ontology.yaml":  exportOntology,
+		"inventory.yaml": exportInventory,
+		"policy.yaml":    exportPolicy,
+	} {
+		doc, err := fn(ctx, pool)
+		if err != nil {
+			return nil, fmt.Errorf("export %s: %w", name, err)
+		}
+		out, err := yaml.Marshal(doc)
+		if err != nil {
+			return nil, fmt.Errorf("marshal %s: %w", name, err)
+		}
+		result[name] = out
+	}
+	return result, nil
+}
+
+func exportOntology(ctx context.Context, pool *Pool) (map[string]any, error) {
+	lifecycles := map[string]any{}
+	rows, err := pool.Query(ctx,
+		`SELECT id, states, default_state, terminal_states, transitions
+		 FROM lifecycle_defs ORDER BY id`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var id, def string
+		var states, terminal []string
+		var transitionsJSON []byte
+		if err := rows.Scan(&id, &states, &def, &terminal, &transitionsJSON); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		var transitions map[string]any
+		if err := json.Unmarshal(transitionsJSON, &transitions); err != nil {
+			rows.Close()
+			return nil, fmt.Errorf("lifecycle %s transitions: %w", id, err)
+		}
+		lifecycles[id] = map[string]any{
+			"states":          states,
+			"default_state":   def,
+			"terminal_states": terminal,
+			"transitions":     transitions,
+		}
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	entityTypes := map[string]any{}
+	rows, err = pool.Query(ctx,
+		`SELECT name, COALESCE(parent_type,''), is_abstract, domain, layer,
+		        COALESCE(description,''), COALESCE(lifecycle_id,''), attribute_schema
+		 FROM entity_types ORDER BY name`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var name, parent, dom, layer, desc, lc string
+		var isAbstract bool
+		var schemaJSON []byte
+		if err := rows.Scan(&name, &parent, &isAbstract, &dom, &layer, &desc, &lc, &schemaJSON); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		et := map[string]any{"domain": dom, "layer": layer}
+		if parent != "" {
+			et["parent"] = parent
+		}
+		if isAbstract {
+			et["abstract"] = true
+		}
+		if desc != "" {
+			et["description"] = desc
+		}
+		if lc != "" {
+			et["lifecycle"] = lc
+		}
+		if len(schemaJSON) > 0 && string(schemaJSON) != "null" {
+			var schema map[string]any
+			if err := json.Unmarshal(schemaJSON, &schema); err == nil && schema != nil {
+				et["attributes"] = schema
+			}
+		}
+		entityTypes[name] = et
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	relTypes := map[string]any{}
+	rows, err = pool.Query(ctx,
+		`SELECT name, COALESCE(inverse,''), source_type, target_type, cardinality,
+		        COALESCE(description,'')
+		 FROM relationship_types ORDER BY name`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var name, inverse, src, tgt, card, desc string
+		if err := rows.Scan(&name, &inverse, &src, &tgt, &card, &desc); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		rt := map[string]any{"source": src, "target": tgt, "cardinality": card}
+		if inverse != "" {
+			rt["inverse"] = inverse
+		}
+		if desc != "" {
+			rt["description"] = desc
+		}
+		relTypes[name] = rt
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	return map[string]any{
+		"version":            1,
+		"lifecycles":         lifecycles,
+		"entity_types":       entityTypes,
+		"relationship_types": relTypes,
+	}, nil
+}
+
+func exportInventory(ctx context.Context, pool *Pool) (map[string]any, error) {
+	var entities []any
+	rows, err := pool.Query(ctx, `
+		SELECT e.slug, e.type, e.name, COALESCE(e.state,''), e.attributes
+		FROM entities e
+		JOIN entity_types et ON et.name = e.type
+		WHERE et.layer <> 'cognition'
+		ORDER BY e.slug`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var slug, typ, name, state string
+		var attrsJSON []byte
+		if err := rows.Scan(&slug, &typ, &name, &state, &attrsJSON); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		e := map[string]any{"slug": slug, "type": typ, "name": name}
+		if state != "" {
+			e["state"] = state
+		}
+		var attrs map[string]any
+		if err := json.Unmarshal(attrsJSON, &attrs); err == nil && len(attrs) > 0 {
+			e["attributes"] = attrs
+		}
+		entities = append(entities, e)
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	var rels []any
+	rows, err = pool.Query(ctx, `
+		SELECT se.slug, te.slug, r.type, r.attributes
+		FROM relationships r
+		JOIN entities se ON se.id = r.source_id
+		JOIN entities te ON te.id = r.target_id
+		WHERE r.valid_to IS NULL
+		ORDER BY r.type, se.slug, te.slug`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var src, tgt, typ string
+		var attrsJSON []byte
+		if err := rows.Scan(&src, &tgt, &typ, &attrsJSON); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		rel := map[string]any{"source": src, "target": tgt, "type": typ}
+		var attrs map[string]any
+		if len(attrsJSON) > 0 && json.Unmarshal(attrsJSON, &attrs) == nil && len(attrs) > 0 {
+			rel["attributes"] = attrs
+		}
+		rels = append(rels, rel)
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	return map[string]any{
+		"version":       1,
+		"entities":      entities,
+		"relationships": rels,
+	}, nil
+}
+
+func exportPolicy(ctx context.Context, pool *Pool) (map[string]any, error) {
+	riskClasses := map[string]any{}
+	rows, err := pool.Query(ctx,
+		`SELECT name, COALESCE(description,''), approval_required, autonomy_allowed
+		 FROM risk_classes ORDER BY name`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var name, desc, approval string
+		var autonomy bool
+		if err := rows.Scan(&name, &desc, &approval, &autonomy); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		rc := map[string]any{"approval_required": approval, "autonomy_allowed": autonomy}
+		if desc != "" {
+			rc["description"] = desc
+		}
+		riskClasses[name] = rc
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	var rules []any
+	rows, err = pool.Query(ctx, `
+		SELECT COALESCE(ar.entity_type,''), ar.action, ar.risk_class,
+		       ar.autonomy_level, COALESCE(se.slug,'')
+		FROM approval_rules ar
+		LEFT JOIN entities se ON se.id = ar.scope_entity
+		ORDER BY COALESCE(ar.entity_type,''), ar.action, COALESCE(se.slug,'')`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var et, action, rc, al, scope string
+		if err := rows.Scan(&et, &action, &rc, &al, &scope); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		rule := map[string]any{"action": action, "risk_class": rc, "autonomy_level": al}
+		if et != "" {
+			rule["entity_type"] = et
+		}
+		if scope != "" {
+			rule["scope_entity"] = scope
+		}
+		rules = append(rules, rule)
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	settings := map[string]any{}
+	rows, err = pool.Query(ctx, `SELECT key, value FROM autonomy_settings ORDER BY key`)
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var k, v string
+		if err := rows.Scan(&k, &v); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		settings[k] = v
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	return map[string]any{
+		"version":           1,
+		"risk_classes":      riskClasses,
+		"approval_rules":    rules,
+		"autonomy_settings": settings,
+	}, nil
+}
--- a/internal/db/integration_test.go
+++ b/internal/db/integration_test.go
@@ -0,0 +1,338 @@
+package db
+
+// Integration tests against a real TimescaleDB. Guarded by
+// OIKOS_TEST_DATABASE_URL — skipped when unset. Run with:
+//
+//   docker compose up -d postgres
+//   OIKOS_TEST_DATABASE_URL="postgres://oikos:oikos_dev@localhost:5432/oikos?sslmode=disable" go test ./internal/db/
+//
+// or `make test-db`. Each run creates a throwaway database and drops it.
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/dtoro/oikos/internal/domain"
+	"github.com/jackc/pgx/v5"
+	"gopkg.in/yaml.v3"
+)
+
+func seedsDir() string { return "../../seeds" }
+
+// newTestPool creates a throwaway database (dropped on cleanup), runs all
+// migrations, and returns a pool connected to it.
+func newTestPool(t *testing.T) *Pool {
+	t.Helper()
+	baseURL := os.Getenv("OIKOS_TEST_DATABASE_URL")
+	if baseURL == "" {
+		t.Skip("OIKOS_TEST_DATABASE_URL not set — skipping integration test")
+	}
+	ctx := context.Background()
+
+	admin, err := pgx.Connect(ctx, baseURL)
+	if err != nil {
+		t.Fatalf("connect admin: %v", err)
+	}
+	dbName := fmt.Sprintf("oikos_test_%08x", rand.Int63())
+	if _, err := admin.Exec(ctx, "CREATE DATABASE "+dbName); err != nil {
+		admin.Close(ctx)
+		t.Fatalf("create test db: %v", err)
+	}
+	admin.Close(ctx)
+
+	testURL := swapDatabase(baseURL, dbName)
+	pool, err := New(ctx, testURL)
+	if err != nil {
+		t.Fatalf("connect test db: %v", err)
+	}
+	t.Cleanup(func() {
+		pool.Close()
+		admin, err := pgx.Connect(ctx, baseURL)
+		if err == nil {
+			admin.Exec(ctx, "DROP DATABASE IF EXISTS "+dbName+" WITH (FORCE)")
+			admin.Close(ctx)
+		}
+	})
+
+	if err := pool.Migrate(ctx); err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	return pool
+}
+
+// swapDatabase replaces the database name in a postgres URL.
+func swapDatabase(url, db string) string {
+	// postgres://user:pass@host:port/dbname?params
+	qi := strings.Index(url, "?")
+	params := ""
+	base := url
+	if qi >= 0 {
+		base, params = url[:qi], url[qi:]
+	}
+	si := strings.LastIndex(base, "/")
+	return base[:si+1] + db + params
+}
+
+func seedAll(t *testing.T, pool *Pool, dir string) {
+	t.Helper()
+	for _, f := range []string{"ontology.yaml", "inventory.yaml", "policy.yaml"} {
+		content, err := os.ReadFile(dir + "/" + f)
+		if err != nil {
+			t.Fatalf("read %s: %v", f, err)
+		}
+		ingestSeedContent(t, pool, f, content)
+	}
+}
+
+func ingestSeedContent(t *testing.T, pool *Pool, name string, content []byte) {
+	t.Helper()
+	ctx := context.Background()
+	err := pool.SeedIngest(ctx, name, content,
+		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
+			var err error
+			switch name {
+			case "ontology.yaml":
+				_, err = IngestOntologySeed(ctx, tx, data)
+			case "inventory.yaml":
+				_, err = IngestInventorySeed(ctx, tx, data)
+			case "policy.yaml":
+				_, err = IngestPolicySeed(ctx, tx, data)
+			}
+			return err
+		})
+	if err != nil {
+		t.Fatalf("ingest %s: %v", name, err)
+	}
+}
+
+func count(t *testing.T, pool *Pool, query string) int {
+	t.Helper()
+	var n int
+	if err := pool.QueryRow(context.Background(), query).Scan(&n); err != nil {
+		t.Fatalf("count %q: %v", query, err)
+	}
+	return n
+}
+
+func TestMigrateIdempotent(t *testing.T) {
+	pool := newTestPool(t)
+	// second run must be a clean no-op
+	if err := pool.Migrate(context.Background()); err != nil {
+		t.Fatalf("second migrate: %v", err)
+	}
+}
+
+func TestSeedIngestIdempotentAndNoDuplicateEdges(t *testing.T) {
+	pool := newTestPool(t)
+	seedAll(t, pool, seedsDir())
+
+	entities := count(t, pool, "SELECT count(*) FROM entities")
+	edges := count(t, pool, "SELECT count(*) FROM relationships WHERE valid_to IS NULL")
+	if entities == 0 || edges == 0 {
+		t.Fatalf("seed produced empty graph: %d entities, %d edges", entities, edges)
+	}
+
+	// Same content → hash no-op
+	seedAll(t, pool, seedsDir())
+	if got := count(t, pool, "SELECT count(*) FROM relationships WHERE valid_to IS NULL"); got != edges {
+		t.Errorf("unchanged re-seed altered edges: %d → %d", edges, got)
+	}
+
+	// Changed content (hash differs) → full re-ingest must NOT duplicate edges
+	// (regression: the old upsert conflicted on valid_from and duplicated all
+	// 144 edges on every re-ingest)
+	content, err := os.ReadFile(seedsDir() + "/inventory.yaml")
+	if err != nil {
+		t.Fatal(err)
+	}
+	touched := append(content, []byte("\n# touched for hash change\n")...)
+	ingestSeedContent(t, pool, "inventory.yaml", touched)
+
+	if got := count(t, pool, "SELECT count(*) FROM relationships WHERE valid_to IS NULL"); got != edges {
+		t.Errorf("touched re-seed duplicated edges: %d → %d", edges, got)
+	}
+	if dup := count(t, pool, `SELECT count(*) FROM (
+			SELECT source_id, target_id, type FROM relationships
+			WHERE valid_to IS NULL GROUP BY 1,2,3 HAVING count(*) > 1) d`); dup != 0 {
+		t.Errorf("%d duplicated current edges", dup)
+	}
+	if got := count(t, pool, "SELECT count(*) FROM entities"); got != entities {
+		t.Errorf("touched re-seed altered entity count: %d → %d", entities, got)
+	}
+}
+
+func TestAbstractTypeRejected(t *testing.T) {
+	pool := newTestPool(t)
+	seedAll(t, pool, seedsDir())
+
+	bad := []byte(`
+version: 1
+entities:
+  - {slug: "machine:ghost", type: machine, name: ghost}
+`)
+	ctx := context.Background()
+	err := pool.SeedIngest(ctx, "inventory.yaml", bad,
+		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
+			_, err := IngestInventorySeed(ctx, tx, data)
+			return err
+		})
+	if !errors.Is(err, domain.ErrAbstractType) {
+		t.Errorf("abstract instantiation = %v, want ErrAbstractType", err)
+	}
+}
+
+func TestEdgeEndpointValidation(t *testing.T) {
+	pool := newTestPool(t)
+	seedAll(t, pool, seedsDir())
+
+	// routes-to requires source ingress-route; a service source must fail
+	bad := []byte(`
+version: 1
+relationships:
+  - {source: "service:gitea", target: "service:caddy", type: routes-to}
+`)
+	ctx := context.Background()
+	err := pool.SeedIngest(ctx, "inventory.yaml", bad,
+		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
+			_, err := IngestInventorySeed(ctx, tx, data)
+			return err
+		})
+	if !errors.Is(err, domain.ErrInvalidEdge) {
+		t.Errorf("bad edge = %v, want ErrInvalidEdge", err)
+	}
+
+	// hosts from a proxmox-host (is-a machine) to an lxc (is-a compute-entity)
+	// must PASS via hierarchy walk — already covered by the seed itself, but
+	// assert an explicit one for clarity
+	good := []byte(`
+version: 1
+relationships:
+  - {source: "host:strong", target: "lxc:jellyfin", type: hosts}
+`)
+	err = pool.SeedIngest(ctx, "inventory.yaml", good,
+		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
+			_, err := IngestInventorySeed(ctx, tx, data)
+			return err
+		})
+	if err != nil {
+		t.Errorf("valid inherited edge rejected: %v", err)
+	}
+}
+
+func TestCardinalityEnforced(t *testing.T) {
+	pool := newTestPool(t)
+	seedAll(t, pool, seedsDir())
+
+	// routes-to is many-to-one: one ingress route cannot point at two services
+	bad := []byte(`
+version: 1
+relationships:
+  - {source: "ingress:git.hubris.network", target: "service:jellyfin", type: routes-to}
+`)
+	ctx := context.Background()
+	err := pool.SeedIngest(ctx, "inventory.yaml", bad,
+		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
+			_, err := IngestInventorySeed(ctx, tx, data)
+			return err
+		})
+	if err == nil || !strings.Contains(err.Error(), "cardinality") {
+		t.Errorf("cardinality violation = %v, want cardinality error", err)
+	}
+}
+
+func TestBlastRadiusTerminatesOnCycles(t *testing.T) {
+	pool := newTestPool(t)
+	seedAll(t, pool, seedsDir())
+
+	// Build a dependency cycle: gitea → caddy → authentik → gitea
+	cycle := []byte(`
+version: 1
+relationships:
+  - {source: "service:gitea", target: "service:caddy", type: depends-on}
+  - {source: "service:caddy", target: "service:authentik", type: depends-on}
+  - {source: "service:authentik", target: "service:gitea", type: depends-on}
+`)
+	ctx := context.Background()
+	err := pool.SeedIngest(ctx, "inventory.yaml", cycle,
+		func(ctx context.Context, tx pgx.Tx, data map[string]any) error {
+			_, err := IngestInventorySeed(ctx, tx, data)
+			return err
+		})
+	if err != nil {
+		t.Fatalf("cycle ingest: %v", err)
+	}
+
+	rows, err := pool.Query(ctx, `
+		SELECT e.slug, b.depth
+		FROM blast_radius((SELECT id FROM entities WHERE slug='service:gitea'), 5,
+		                  ARRAY['depends-on']) b
+		JOIN entities e ON e.id = b.entity_id ORDER BY b.depth`)
+	if err != nil {
+		t.Fatalf("blast_radius: %v", err)
+	}
+	defer rows.Close()
+	got := map[string]int{}
+	for rows.Next() {
+		var slug string
+		var depth int
+		if err := rows.Scan(&slug, &depth); err != nil {
+			t.Fatal(err)
+		}
+		got[slug] = depth
+	}
+	want := map[string]int{"service:gitea": 0, "service:caddy": 1, "service:authentik": 2}
+	for slug, depth := range want {
+		if got[slug] != depth {
+			t.Errorf("blast_radius[%s] = %d, want %d (full: %v)", slug, got[slug], depth, got)
+		}
+	}
+	if len(got) != len(want) {
+		t.Errorf("blast_radius returned %d nodes, want %d: %v", len(got), len(want), got)
+	}
+}
+
+// TestExportRoundTripStable: export → ingest into a fresh DB → export again
+// must yield byte-identical YAML (the canonical-form fixpoint, plan D6).
+func TestExportRoundTripStable(t *testing.T) {
+	pool := newTestPool(t)
+	seedAll(t, pool, seedsDir())
+	ctx := context.Background()
+
+	export1, err := ExportToYAML(ctx, pool)
+	if err != nil {
+		t.Fatalf("export 1: %v", err)
+	}
+	for name, content := range export1 {
+		var doc map[string]any
+		if err := yaml.Unmarshal(content, &doc); err != nil {
+			t.Fatalf("export %s is not valid YAML: %v", name, err)
+		}
+	}
+
+	pool2 := newTestPool(t)
+	for _, name := range []string{"ontology.yaml", "inventory.yaml", "policy.yaml"} {
+		ingestSeedContent(t, pool2, name, export1[name])
+	}
+	export2, err := ExportToYAML(ctx, pool2)
+	if err != nil {
+		t.Fatalf("export 2: %v", err)
+	}
+	for _, name := range []string{"ontology.yaml", "inventory.yaml", "policy.yaml"} {
+		if !bytes.Equal(export1[name], export2[name]) {
+			t.Errorf("%s round-trip not byte-stable (len %d vs %d)",
+				name, len(export1[name]), len(export2[name]))
+		}
+	}
+
+	// sanity: exported inventory carries the real graph, not a stub
+	// (regression: export used to write 11-byte "version: 1" stubs)
+	if len(export1["inventory.yaml"]) < 1000 {
+		t.Errorf("inventory export suspiciously small: %d bytes", len(export1["inventory.yaml"]))
+	}
+}
--- a/internal/db/pool.go
+++ b/internal/db/pool.go
@@ -39,11 +39,27 @@ func New(ctx context.Context, databaseURL string) (*Pool, error) {
 	return &Pool{pool}, nil
 }

+// migrationLockKey is the advisory-lock key serializing migration runs —
+// two concurrent `oikos migrate` invocations must not interleave DDL.
+const migrationLockKey = 0x01c05e5
+
 // Migrate runs all embedded forward migrations in order.
-// Uses a schema_migrations table to track applied versions.
+// Uses a schema_migrations table to track applied versions. The whole run
+// happens on one connection holding a session advisory lock.
 func (p *Pool) Migrate(ctx context.Context) error {
+	conn, err := p.Acquire(ctx)
+	if err != nil {
+		return fmt.Errorf("acquire migration conn: %w", err)
+	}
+	defer conn.Release()
+
+	if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock($1)", migrationLockKey); err != nil {
+		return fmt.Errorf("acquire migration lock: %w", err)
+	}
+	defer conn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", migrationLockKey)
+
 	// Create tracking table if not exists
-	_, err := p.Exec(ctx, `
+	_, err = conn.Exec(ctx, `
 		CREATE TABLE IF NOT EXISTS schema_migrations (
 			version INT PRIMARY KEY,
 			applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
@@ -76,7 +92,7 @@ func (p *Pool) Migrate(ctx context.Context) error {

 		// Check if already applied
 		var applied int
-		err := p.QueryRow(ctx,
+		err := conn.QueryRow(ctx,
 			"SELECT COUNT(*) FROM schema_migrations WHERE version = $1", version).Scan(&applied)
 		if err != nil {
 			return fmt.Errorf("check migration %d: %w", version, err)
@@ -99,12 +115,12 @@ func (p *Pool) Migrate(ctx context.Context) error {
 			if stmt == "" {
 				continue
 			}
-			_, err := p.Exec(ctx, stmt)
+			_, err := conn.Exec(ctx, stmt)
 			if err != nil {
 				return fmt.Errorf("exec migration %s stmt %d: %w", fname, i+1, err)
 			}
 		}
-		_, err = p.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version)
+		_, err = conn.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", version)
 		if err != nil {
 			return fmt.Errorf("record migration %d: %w", version, err)
 		}
--- a/internal/db/seed.go
+++ b/internal/db/seed.go
@@ -3,13 +3,11 @@ package db
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
-	"time"

-	"github.com/dtoro/oikos/internal/domain"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
-	"gopkg.in/yaml.v3"
 )

 // SeedResult holds counts from a seed ingest operation.
@@ -83,11 +81,20 @@ func IngestOntologySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*S
 }

 // IngestInventorySeed ingests seeds/inventory.yaml into the DB.
+// Every entity and edge is validated against the ontology (abstract types
+// rejected, lifecycle states checked, relationship endpoints hierarchy-
+// validated, cardinality enforced) — a violating seed rolls back atomically.
 func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, error) {
 	r := &SeedResult{}

+	tree, err := LoadTypeTree(ctx, tx)
+	if err != nil {
+		return nil, fmt.Errorf("load type tree: %w", err)
+	}
+
 	// Entities
 	entities, _ := data["entities"].([]any)
+	entityTypes := make(map[string]string) // slug -> type, for edge validation
 	for _, raw := range entities {
 		eMap, ok := raw.(map[string]any)
 		if !ok {
@@ -99,7 +106,14 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*
 		state, _ := eMap["state"].(string)
 		attrs := eMap["attributes"]

-		// Generate UUIDv7 for new entities, or find existing by slug
+		if err := tree.ValidateEntity(typeName, state); err != nil {
+			return nil, fmt.Errorf("entity %s: %w", slug, err)
+		}
+		if state == "" {
+			state = tree.DefaultState(typeName)
+		}
+		entityTypes[slug] = typeName
+
 		entityID, err := getOrCreateEntityID(ctx, tx, slug)
 		if err != nil {
 			return nil, fmt.Errorf("entity %s: %w", slug, err)
@@ -118,7 +132,8 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*
 		r.Entities++
 	}

-	// Relationships
+	// Relationships — upsert against the current-edge partial unique index
+	// (migration 007) so re-ingest never duplicates edges.
 	rels, _ := data["relationships"].([]any)
 	for _, raw := range rels {
 		relMap, ok := raw.(map[string]any)
@@ -139,12 +154,32 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*
 			return nil, fmt.Errorf("rel to %s: %w", target, err)
 		}

+		srcType := entityTypes[source]
+		tgtType := entityTypes[target]
+		if srcType == "" || tgtType == "" { // entity pre-existing in DB, not in this seed
+			if srcType == "" {
+				srcType, err = getEntityTypeBySlug(ctx, tx, source)
+				if err != nil {
+					return nil, err
+				}
+			}
+			if tgtType == "" {
+				tgtType, err = getEntityTypeBySlug(ctx, tx, target)
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
+		if err := tree.ValidateEdge(relType, srcType, tgtType); err != nil {
+			return nil, fmt.Errorf("rel %s→%s: %w", source, target, err)
+		}
+
 		attrsBytes, _ := json.Marshal(attrs)
 		_, err = tx.Exec(ctx,
 			`INSERT INTO relationships (source_id, target_id, type, attributes, valid_from, valid_to)
 			 VALUES ($1, $2, $3, $4, now(), NULL)
-			 ON CONFLICT (source_id, target_id, type, valid_from)
-			 DO UPDATE SET attributes = $4`,
+			 ON CONFLICT (source_id, target_id, type) WHERE valid_to IS NULL
+			 DO UPDATE SET attributes = EXCLUDED.attributes`,
 			sourceID, targetID, relType, string(attrsBytes))
 		if err != nil {
 			return nil, fmt.Errorf("rel %s→%s %s: %w", source, target, relType, err)
@@ -152,6 +187,10 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*
 		r.Relationships++
 	}

+	if err := ValidateCardinality(ctx, tx); err != nil {
+		return nil, err
+	}
+
 	return r, nil
 }

@@ -231,15 +270,29 @@ func IngestPolicySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*See

 // --- Helpers ---

-// getOrCreateEntityID returns the UUID for a slug, generating a new UUIDv7 if not found.
+// getOrCreateEntityID returns the UUID for a slug, generating a new
+// time-ordered UUIDv7 if the slug doesn't exist yet (ADR-0005).
 func getOrCreateEntityID(ctx context.Context, tx pgx.Tx, slug string) (uuid.UUID, error) {
 	var id uuid.UUID
 	err := tx.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", slug).Scan(&id)
-	if err == nil {
+	switch {
+	case err == nil:
 		return id, nil
+	case errors.Is(err, pgx.ErrNoRows):
+		return uuid.NewV7()
+	default:
+		return uuid.Nil, fmt.Errorf("lookup slug %s: %w", slug, err)
 	}
-	// Generate a time-ordered UUID (using uuid.New for now; UUIDv7 in production)
-	return uuid.New(), nil
+}
+
+// getEntityTypeBySlug resolves a slug to its entity type name.
+func getEntityTypeBySlug(ctx context.Context, tx pgx.Tx, slug string) (string, error) {
+	var t string
+	err := tx.QueryRow(ctx, "SELECT type FROM entities WHERE slug = $1", slug).Scan(&t)
+	if err != nil {
+		return "", fmt.Errorf("resolve type of %s: %w", slug, err)
+	}
+	return t, nil
 }

 // getEntityIDBySlug resolves a slug to its UUID.
@@ -338,49 +391,3 @@ func keysOf(m map[string]map[string]any) []string {
 	return keys
 }

-// ExportToYAML regenerates the three seed YAMLs from the DB (for DR / version control, D6).
-func ExportToYAML(ctx context.Context, pool *Pool) (map[string][]byte, error) {
-	result := make(map[string][]byte)
-
-	// Export ontology
-	onto, err := exportOntology(ctx, pool)
-	if err != nil {
-		return nil, fmt.Errorf("export ontology: %w", err)
-	}
-	result["ontology.yaml"], _ = yaml.Marshal(onto)
-
-	// Export inventory
-	inv, err := exportInventory(ctx, pool)
-	if err != nil {
-		return nil, fmt.Errorf("export inventory: %w", err)
-	}
-	result["inventory.yaml"], _ = yaml.Marshal(inv)
-
-	// Export policy
-	pol, err := exportPolicy(ctx, pool)
-	if err != nil {
-		return nil, fmt.Errorf("export policy: %w", err)
-	}
-	result["policy.yaml"], _ = yaml.Marshal(pol)
-
-	return result, nil
-}
-
-func exportOntology(ctx context.Context, pool *Pool) (map[string]any, error) {
-	// TODO: implement full export from DB
-	return map[string]any{"version": 1}, nil
-}
-
-func exportInventory(ctx context.Context, pool *Pool) (map[string]any, error) {
-	// TODO: implement full export from DB
-	return map[string]any{"version": 1}, nil
-}
-
-func exportPolicy(ctx context.Context, pool *Pool) (map[string]any, error) {
-	// TODO: implement full export from DB
-	return map[string]any{"version": 1}, nil
-}
-
-// Unused import suppression for domain (will be needed when we add more logic)
-var _ = domain.Entity{}
-var _ = time.Now
--- a/internal/db/splitsql_test.go
+++ b/internal/db/splitsql_test.go
@@ -0,0 +1,53 @@
+package db
+
+import (
+	"strings"
+	"testing"
+)
+
+func nonEmpty(stmts []string) []string {
+	var out []string
+	for _, s := range stmts {
+		if strings.TrimSpace(s) != "" {
+			out = append(out, s)
+		}
+	}
+	return out
+}
+
+func TestSplitSQLBasic(t *testing.T) {
+	stmts := nonEmpty(splitSQL("CREATE TABLE a (id int); CREATE TABLE b (id int);"))
+	if len(stmts) != 2 {
+		t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts)
+	}
+}
+
+func TestSplitSQLDollarQuotedFunction(t *testing.T) {
+	sql := `CREATE FUNCTION f() RETURNS int AS $$
+  SELECT 1; SELECT 2;
+$$ LANGUAGE sql;
+CREATE TABLE t (id int);`
+	stmts := nonEmpty(splitSQL(sql))
+	if len(stmts) != 2 {
+		t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts)
+	}
+	if !strings.Contains(stmts[0], "SELECT 1; SELECT 2;") {
+		t.Errorf("dollar-quoted body was split: %q", stmts[0])
+	}
+}
+
+func TestSplitSQLTaggedDollarQuote(t *testing.T) {
+	sql := `DO $body$ BEGIN PERFORM 1; END $body$;SELECT 1;`
+	stmts := nonEmpty(splitSQL(sql))
+	if len(stmts) != 2 {
+		t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts)
+	}
+}
+
+func TestSplitSQLSemicolonInComment(t *testing.T) {
+	sql := "-- comment with ; semicolon\nCREATE TABLE t (id int); -- trailing; note\nSELECT 1;"
+	stmts := nonEmpty(splitSQL(sql))
+	if len(stmts) != 2 {
+		t.Fatalf("got %d statements, want 2: %#v", len(stmts), stmts)
+	}
+}
--- a/internal/db/typetree.go
+++ b/internal/db/typetree.go
@@ -0,0 +1,143 @@
+package db
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/dtoro/oikos/internal/ontology"
+	"github.com/jackc/pgx/v5"
+)
+
+// LoadTypeTree loads the ontology meta-schema from the DB for validation.
+// Called within the same transaction as an ingest so it sees just-ingested
+// types.
+func LoadTypeTree(ctx context.Context, tx pgx.Tx) (*ontology.TypeTree, error) {
+	t := &ontology.TypeTree{
+		Types:      map[string]ontology.TypeInfo{},
+		RelTypes:   map[string]ontology.RelTypeInfo{},
+		Lifecycles: map[string]ontology.LifecycleInfo{},
+	}
+
+	rows, err := tx.Query(ctx,
+		`SELECT name, COALESCE(parent_type,''), is_abstract, COALESCE(lifecycle_id,'')
+		 FROM entity_types`)
+	if err != nil {
+		return nil, fmt.Errorf("load entity_types: %w", err)
+	}
+	for rows.Next() {
+		var name string
+		var info ontology.TypeInfo
+		if err := rows.Scan(&name, &info.Parent, &info.IsAbstract, &info.LifecycleID); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		t.Types[name] = info
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	rows, err = tx.Query(ctx,
+		`SELECT name, source_type, target_type, cardinality FROM relationship_types`)
+	if err != nil {
+		return nil, fmt.Errorf("load relationship_types: %w", err)
+	}
+	for rows.Next() {
+		var name string
+		var info ontology.RelTypeInfo
+		if err := rows.Scan(&name, &info.SourceType, &info.TargetType, &info.Cardinality); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		t.RelTypes[name] = info
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	rows, err = tx.Query(ctx,
+		`SELECT id, states, default_state FROM lifecycle_defs`)
+	if err != nil {
+		return nil, fmt.Errorf("load lifecycle_defs: %w", err)
+	}
+	for rows.Next() {
+		var id, def string
+		var states []string
+		if err := rows.Scan(&id, &states, &def); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		set := make(map[string]bool, len(states))
+		for _, s := range states {
+			set[s] = true
+		}
+		t.Lifecycles[id] = ontology.LifecycleInfo{States: set, DefaultState: def}
+	}
+	rows.Close()
+	if rows.Err() != nil {
+		return nil, rows.Err()
+	}
+
+	return t, nil
+}
+
+// ValidateCardinality checks all CURRENT edges against their relationship
+// type's declared cardinality. Runs inside the ingest transaction so a
+// violating seed rolls back atomically.
+func ValidateCardinality(ctx context.Context, tx pgx.Tx) error {
+	// one-to-one / many-to-one: a source may have at most one outgoing
+	// current edge of the type.
+	rows, err := tx.Query(ctx, `
+		SELECT r.type, se.slug, count(*)
+		FROM relationships r
+		JOIN relationship_types rt ON rt.name = r.type
+		JOIN entities se ON se.id = r.source_id
+		WHERE r.valid_to IS NULL AND rt.cardinality IN ('one-to-one','many-to-one')
+		GROUP BY r.type, se.slug HAVING count(*) > 1`)
+	if err != nil {
+		return err
+	}
+	violations, err := collectViolations(rows, "source")
+	if err != nil {
+		return err
+	}
+
+	// one-to-one / one-to-many: a target may have at most one incoming
+	// current edge of the type.
+	rows, err = tx.Query(ctx, `
+		SELECT r.type, te.slug, count(*)
+		FROM relationships r
+		JOIN relationship_types rt ON rt.name = r.type
+		JOIN entities te ON te.id = r.target_id
+		WHERE r.valid_to IS NULL AND rt.cardinality IN ('one-to-one','one-to-many')
+		GROUP BY r.type, te.slug HAVING count(*) > 1`)
+	if err != nil {
+		return err
+	}
+	v2, err := collectViolations(rows, "target")
+	if err != nil {
+		return err
+	}
+	violations = append(violations, v2...)
+
+	if len(violations) > 0 {
+		return fmt.Errorf("cardinality violations: %v", violations)
+	}
+	return nil
+}
+
+func collectViolations(rows pgx.Rows, side string) ([]string, error) {
+	defer rows.Close()
+	var out []string
+	for rows.Next() {
+		var relType, slug string
+		var n int
+		if err := rows.Scan(&relType, &slug, &n); err != nil {
+			return nil, err
+		}
+		out = append(out, fmt.Sprintf("%s %s=%s (%d edges)", relType, side, slug, n))
+	}
+	return out, rows.Err()
+}