diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 0fd0b112..240e25e9 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -57,12 +57,12 @@ internal/ All Go packages the phased refactor; core may not import adapters, enforced by depguard (verified Phase 9 audit) adapters/ Ports' implementations: postgres/ (pool, migrations, seeds, - sqlcgen, repos — EntityRepo, RelRepo, OntologyRepo, - EntityReader, SignalRepo, MetricsRepo, etc.), - ssh/ (CommandExecutor via actuator), - remote/ (TargetResolver via internal/remote), - probes/ (Checker implementations per probe kind: - HTTP, TCP, ping, DNS, SSH) + incl. knowledge seed ingest, sqlcgen, repos — EntityRepo, + RelRepo, OntologyRepo, SeedRepo, EntityReader, SignalRepo, + MetricsRepo, etc.), ssh/ (CommandExecutor via actuator + + Provisioner for pct/qm), remote/ (TargetResolver via + internal/remote), probes/ (Checker implementations per + probe kind: HTTP, TCP, ping, DNS, SSH) httpapi/ REST server (OpenAPI-generated) — driving adapter mcp/ MCP tool implementations — driving adapter scheduler/ Observe loop, coverage sweep — driving adapter (moves @@ -72,7 +72,6 @@ internal/ All Go packages remote/ Target resolution — consumed by adapters/remote audit/ Audit report helpers observability/ Event/Audit recorder helpers - knowledge/ Knowledge YAML seed ingestion api/openapi.yaml API contract — the source of truth for endpoints migrations/ Forward-only SQL migrations (TimescaleDB) seeds/ Bootstrap YAML: ontology, inventory, policy, knowledge diff --git a/VERSION b/VERSION index 85e60ed1..cd46610f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.34.0 +0.34.1 diff --git a/cmd/oikos/main.go b/cmd/oikos/main.go index ecf39b49..641ce2e5 100644 --- a/cmd/oikos/main.go +++ b/cmd/oikos/main.go @@ -13,14 +13,15 @@ import ( "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/core/app" + "github.com/dtoro/oikos/internal/core/ports" "github.com/dtoro/oikos/internal/adapters/postgres" + "github.com/dtoro/oikos/internal/adapters/remote" + "github.com/dtoro/oikos/internal/adapters/ssh" "github.com/dtoro/oikos/internal/execworker" "github.com/dtoro/oikos/internal/httpapi" - "github.com/dtoro/oikos/internal/knowledge" "github.com/dtoro/oikos/internal/observability" "github.com/dtoro/oikos/internal/scheduler" "github.com/dtoro/oikos/internal/secrets" - "github.com/jackc/pgx/v5" ) var schedulerRunner = scheduler.RunnerForMain() @@ -111,15 +112,9 @@ func main() { go schedulerRunner(ctx, pool, cfg) go execWorkerRunner(ctx, pool, cfg) - // Build composition-root dependencies (ADR 0016). - entityRepo := db.NewEntityRepo(pool) - onto := db.NewOntologyRepo(pool, time.Minute) - readModels := db.NewEntityReader(pool) - entities := app.NewEntityService(entityRepo, onto) - relService := app.NewRelationshipService(db.NewRelRepo(pool), onto) - slog.Info("all: starting api with scheduler + execution-worker in background") - if err := httpapi.ListenAndServe(ctx, pool, cfg, entities, entityRepo, readModels, relService); err != nil { + svc := buildAPIServices(pool) + if err := httpapi.ListenAndServe(ctx, pool, cfg, svc.entities, svc.entityRepo, svc.readModels, svc.relService, svc.provisioning, svc.seeds); err != nil { slog.Error("api failed", "error", err) os.Exit(1) } @@ -177,6 +172,9 @@ func runMigrate(ctx context.Context, cfg config.Config) error { return nil } +// runSeed ingests the seed YAMLs via SeedService (Phase 7): the CLI is +// an adapter over the service; ordering, idempotency (content-hash +// no-op), and per-file logging live in the service/repository pair. func runSeed(ctx context.Context, cfg config.Config) error { pool, err := db.New(ctx, cfg.DatabaseURL) if err != nil { @@ -194,96 +192,7 @@ func runSeed(ctx context.Context, cfg config.Config) error { seedsDir = "seeds" } - // Ingest ontology seed - ontoContent, err := os.ReadFile(seedsDir + "/ontology.yaml") - if err != nil { - return fmt.Errorf("read ontology seed: %w", err) - } - err = pool.SeedIngest(ctx, "ontology.yaml", ontoContent, - func(ctx context.Context, tx pgx.Tx, data map[string]any) error { - r, err := db.IngestOntologySeed(ctx, tx, data) - if err != nil { - return err - } - slog.Info("ontology ingested", - "lifecycles", r.Lifecycles, - "entity_types", r.EntityTypes, - "relationship_types", r.RelationshipTypes) - return nil - }) - if err != nil { - return err - } - - // Ingest inventory seed - invContent, err := os.ReadFile(seedsDir + "/inventory.yaml") - if err != nil { - return fmt.Errorf("read inventory seed: %w", err) - } - err = pool.SeedIngest(ctx, "inventory.yaml", invContent, - func(ctx context.Context, tx pgx.Tx, data map[string]any) error { - r, err := db.IngestInventorySeed(ctx, tx, data) - if err != nil { - return err - } - slog.Info("inventory ingested", - "entities", r.Entities, - "relationships", r.Relationships) - return nil - }) - if err != nil { - return err - } - - // Ingest policy seed - polContent, err := os.ReadFile(seedsDir + "/policy.yaml") - if err != nil { - return fmt.Errorf("read policy seed: %w", err) - } - err = pool.SeedIngest(ctx, "policy.yaml", polContent, - func(ctx context.Context, tx pgx.Tx, data map[string]any) error { - r, err := db.IngestPolicySeed(ctx, tx, data) - if err != nil { - return err - } - slog.Info("policy ingested", - "risk_classes", r.RiskClasses, - "approval_rules", r.ApprovalRules, - "autonomy_settings", r.AutonomySettings) - return nil - }) - if err != nil { - return err - } - - // Ingest knowledge seed (documents, investigations, runbooks) - knContent, err := os.ReadFile(seedsDir + "/knowledge.yaml") - if err != nil { - if os.IsNotExist(err) { - slog.Info("knowledge seed not found, skipping") - } else { - return fmt.Errorf("read knowledge seed: %w", err) - } - } else { - err = pool.SeedIngest(ctx, "knowledge.yaml", knContent, - func(ctx context.Context, tx pgx.Tx, data map[string]any) error { - r, err := knowledge.Ingest(ctx, tx, data) - if err != nil { - return err - } - slog.Info("knowledge ingested", - "documents", r.Documents, - "investigations", r.Investigations, - "runbooks", r.Runbooks) - return nil - }) - if err != nil { - return err - } - } - - slog.Info("seed ingest complete") - return nil + return app.NewSeedService(db.NewSeedRepo(pool)).Ingest(ctx, seedsDir) } func runAPI(ctx context.Context, cfg config.Config) error { @@ -297,18 +206,48 @@ func runAPI(ctx context.Context, cfg config.Config) error { return fmt.Errorf("migrations: %w", err) } - // Composition root — build the service dependencies (ADR 0016, plan §3.5). + svc := buildAPIServices(pool) + err = httpapi.ListenAndServe(ctx, pool, cfg, svc.entities, svc.entityRepo, svc.readModels, svc.relService, svc.provisioning, svc.seeds) + if err == http.ErrServerClosed { + return nil + } + return err +} + +// apiServices is the composition-root dependency set for the API surface +// (ADR 0016, plan §3.5): repositories, the ssh executor and provisioner, +// the target resolver, and the use-case services built on them. +type apiServices struct { + entities *app.EntityService + entityRepo *db.EntityRepo + readModels ports.ReadModels + relService *app.RelationshipService + provisioning *app.ProvisioningService + seeds *app.SeedService +} + +func buildAPIServices(pool *db.Pool) apiServices { entityRepo := db.NewEntityRepo(pool) onto := db.NewOntologyRepo(pool, time.Minute) readModels := db.NewEntityReader(pool) entities := app.NewEntityService(entityRepo, onto) relService := app.NewRelationshipService(db.NewRelRepo(pool), onto) - err = httpapi.ListenAndServe(ctx, pool, cfg, entities, entityRepo, readModels, relService) - if err == http.ErrServerClosed { - return nil + executor := ssh.NewExecutor(ssh.FileSignerSource(), 5*time.Minute) + provisioner := ssh.NewProvisioner(executor) + resolver := remote.NewResolver(pool) + provisioning := app.NewProvisioningService(provisioner, resolver, entityRepo, db.NewRelRepo(pool)) + + seeds := app.NewSeedService(db.NewSeedRepo(pool)) + + return apiServices{ + entities: entities, + entityRepo: entityRepo, + readModels: readModels, + relService: relService, + provisioning: provisioning, + seeds: seeds, } - return err } func runWithPool(ctx context.Context, cfg config.Config, name string, fn func(context.Context, *db.Pool, config.Config)) { @@ -329,13 +268,15 @@ func runWithPool(ctx context.Context, cfg config.Config, name string, fn func(co func runSecret(ctx context.Context, cfg config.Config) { if len(os.Args) < 3 { - fmt.Fprintln(os.Stderr, "usage: oikos secret ") + fmt.Fprintln(os.Stderr, "usage: oikos secret ") os.Exit(1) } sub := os.Args[2] - // For get/set/list: use Infisical directly + // get/set/list go through SecretsService (Phase 7); the CLI is an + // adapter over the service. verify/audit/migrate/export-sops remain + // backend-specific diagnostics. switch sub { case "get": if len(os.Args) < 4 { @@ -343,8 +284,8 @@ func runSecret(ctx context.Context, cfg config.Config) { os.Exit(1) } key := os.Args[3] - backend := newInfisicalBackendOrFail(cfg) - val, err := backend.Get(ctx, key) + svc := app.NewSecretsService(newInfisicalBackendOrFail(cfg)) + val, err := svc.Get(ctx, key) if err != nil { slog.Error("secret get", "key", key, "error", err) os.Exit(1) @@ -358,16 +299,16 @@ func runSecret(ctx context.Context, cfg config.Config) { } key := os.Args[3] value := os.Args[4] - backend := newInfisicalBackendOrFail(cfg) - if err := backend.Set(ctx, key, value); err != nil { + svc := app.NewSecretsService(newInfisicalBackendOrFail(cfg)) + if err := svc.Set(ctx, key, value); err != nil { slog.Error("secret set", "key", key, "error", err) os.Exit(1) } fmt.Printf("stored: %s\n", key) case "list": - backend := newInfisicalBackendOrFail(cfg) - keys, err := backend.List(ctx) + svc := app.NewSecretsService(newInfisicalBackendOrFail(cfg)) + keys, err := svc.List(ctx) if err != nil { slog.Error("secret list", "error", err) os.Exit(1) @@ -544,6 +485,8 @@ func runSecretLegacy(ctx context.Context, cfg config.Config, sub string) { } } +// runExport regenerates the seed YAMLs from the DB via SeedService and +// writes them into the seeds dir (DR / version control). func runExport(ctx context.Context, cfg config.Config) error { pool, err := db.New(ctx, cfg.DatabaseURL) if err != nil { @@ -551,7 +494,7 @@ func runExport(ctx context.Context, cfg config.Config) error { } defer pool.Close() - exports, err := db.ExportToYAML(ctx, pool) + exports, err := app.NewSeedService(db.NewSeedRepo(pool)).Export(ctx) if err != nil { return err } diff --git a/internal/adapters/postgres/repositories.go b/internal/adapters/postgres/repositories.go index 9c5e15e4..125c4623 100644 --- a/internal/adapters/postgres/repositories.go +++ b/internal/adapters/postgres/repositories.go @@ -259,10 +259,10 @@ func (r *EntityRepo) Create(ctx context.Context, in ports.EntityCreateInput) (do } created, err := scanDomainEntity(tx.QueryRow(ctx, ` - INSERT INTO entities (id, slug, type, name, state, attributes) - VALUES ($1, $2, $3, $4, $5, $6) + INSERT INTO entities (id, slug, type, name, state, attributes, enrolled_at) + VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING `+entityFullCols, - mustUUID(e.ID), e.Slug, e.Type, e.Name, state, attrsJSON)) + mustUUID(e.ID), e.Slug, e.Type, e.Name, state, attrsJSON, in.EnrolledAt)) if err != nil { if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") { return domain.Entity{}, errors.Join(domain.ErrAlreadyExists, err) diff --git a/internal/knowledge/seed.go b/internal/adapters/postgres/seedrepo.go similarity index 51% rename from internal/knowledge/seed.go rename to internal/adapters/postgres/seedrepo.go index 34299526..1fcf7978 100644 --- a/internal/knowledge/seed.go +++ b/internal/adapters/postgres/seedrepo.go @@ -1,35 +1,132 @@ -package knowledge +package db import ( "context" - "crypto/sha256" - "encoding/hex" "encoding/json" "fmt" + "github.com/dtoro/oikos/internal/core/ports" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) -type SeedResult struct { +// SeedRepo implements ports.SeedRepository over the postgres pool. Each +// Ingest method wraps Pool.SeedIngest (hash no-op + one transaction + +// seed_versions recording) around the file's ingest functions; the +// knowledge ingest moved here from internal/knowledge when SeedService +// absorbed seeding (Phase 7) — raw SQL over the open transaction belongs +// in the adapter. +type SeedRepo struct { + pool *Pool +} + +var _ ports.SeedRepository = (*SeedRepo)(nil) + +// NewSeedRepo builds the seed repository. +func NewSeedRepo(pool *Pool) *SeedRepo { return &SeedRepo{pool: pool} } + +// IngestOntology ingests seeds/ontology.yaml (lifecycles, entity types, +// relationship types) — one transaction, no-op on unchanged hash. +func (r *SeedRepo) IngestOntology(ctx context.Context, filename string, content []byte) (ports.SeedCounts, bool, error) { + var counts ports.SeedCounts + applied := false + err := r.pool.SeedIngest(ctx, filename, content, func(ctx context.Context, tx pgx.Tx, data map[string]any) error { + res, err := IngestOntologySeed(ctx, tx, data) + if err != nil { + return err + } + counts.Lifecycles = res.Lifecycles + counts.EntityTypes = res.EntityTypes + counts.RelationshipTypes = res.RelationshipTypes + applied = true + return nil + }) + return counts, applied, err +} + +// IngestInventory ingests seeds/inventory.yaml (entities, relationships, +// derived default checks) with ontology validation — a violating seed +// rolls back atomically. +func (r *SeedRepo) IngestInventory(ctx context.Context, filename string, content []byte) (ports.SeedCounts, bool, error) { + var counts ports.SeedCounts + applied := false + err := r.pool.SeedIngest(ctx, filename, content, func(ctx context.Context, tx pgx.Tx, data map[string]any) error { + res, err := IngestInventorySeed(ctx, tx, data) + if err != nil { + return err + } + counts.Entities = res.Entities + counts.Relationships = res.Relationships + counts.Checks = res.Checks + applied = true + return nil + }) + return counts, applied, err +} + +// IngestPolicy ingests seeds/policy.yaml (risk classes, approval rules, +// autonomy settings). +func (r *SeedRepo) IngestPolicy(ctx context.Context, filename string, content []byte) (ports.SeedCounts, bool, error) { + var counts ports.SeedCounts + applied := false + err := r.pool.SeedIngest(ctx, filename, content, func(ctx context.Context, tx pgx.Tx, data map[string]any) error { + res, err := IngestPolicySeed(ctx, tx, data) + if err != nil { + return err + } + counts.RiskClasses = res.RiskClasses + counts.ApprovalRules = res.ApprovalRules + counts.AutonomySettings = res.AutonomySettings + applied = true + return nil + }) + return counts, applied, err +} + +// IngestKnowledge ingests seeds/knowledge.yaml (documents, +// investigations, runbooks) as knowledge entities linked into the graph. +func (r *SeedRepo) IngestKnowledge(ctx context.Context, filename string, content []byte) (ports.SeedCounts, bool, error) { + var counts ports.SeedCounts + applied := false + err := r.pool.SeedIngest(ctx, filename, content, func(ctx context.Context, tx pgx.Tx, data map[string]any) error { + res, err := IngestKnowledgeSeed(ctx, tx, data) + if err != nil { + return err + } + counts.Documents = res.Documents + counts.Investigations = res.Investigations + counts.Runbooks = res.Runbooks + applied = true + return nil + }) + return counts, applied, err +} + +// Export regenerates the three structural seed YAMLs from the DB. +func (r *SeedRepo) Export(ctx context.Context) (map[string][]byte, error) { + return ExportToYAML(ctx, r.pool) +} + +// KnowledgeSeedResult holds counts from the knowledge seed ingest. +type KnowledgeSeedResult struct { Documents int Investigations int Runbooks int } -func contentHash(s string) string { - h := sha256.Sum256([]byte(s)) - return hex.EncodeToString(h[:]) -} - -func Ingest(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, error) { - r := &SeedResult{} +// IngestKnowledgeSeed ingests the knowledge seed (documents, +// investigations, runbooks) into knowledge entities plus their graph +// edges, inside the caller's ingest transaction. Absorbed from +// internal/knowledge (Phase 7): the logic is seed ingest, its home is the +// seed repository. +func IngestKnowledgeSeed(ctx context.Context, tx pgx.Tx, data map[string]any) (*KnowledgeSeedResult, error) { + r := &KnowledgeSeedResult{} docs, _ := data["documents"].([]any) for _, raw := range docs { d, _ := raw.(map[string]any) if err := ingestDocument(ctx, tx, d); err != nil { - return nil, fmt.Errorf("document %v: %w", str(d, "slug"), err) + return nil, fmt.Errorf("document %v: %w", seedStr(d, "slug"), err) } r.Documents++ } @@ -38,7 +135,7 @@ func Ingest(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, e for _, raw := range invs { m, _ := raw.(map[string]any) if err := ingestInvestigation(ctx, tx, m); err != nil { - return nil, fmt.Errorf("investigation %v: %w", str(m, "slug"), err) + return nil, fmt.Errorf("investigation %v: %w", seedStr(m, "slug"), err) } r.Investigations++ } @@ -47,7 +144,7 @@ func Ingest(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, e for _, raw := range rbs { m, _ := raw.(map[string]any) if err := ingestRunbook(ctx, tx, m); err != nil { - return nil, fmt.Errorf("runbook %v: %w", str(m, "slug"), err) + return nil, fmt.Errorf("runbook %v: %w", seedStr(m, "slug"), err) } r.Runbooks++ } @@ -55,12 +152,12 @@ func Ingest(ctx context.Context, tx pgx.Tx, data map[string]any) (*SeedResult, e return r, nil } -func str(m map[string]any, key string) string { +func seedStr(m map[string]any, key string) string { s, _ := m[key].(string) return s } -func strSlice(m map[string]any, key string) []string { +func seedStrSlice(m map[string]any, key string) []string { raw, _ := m[key].([]any) var out []string for _, v := range raw { @@ -71,17 +168,12 @@ func strSlice(m map[string]any, key string) []string { return out } -func mapVal(m map[string]any, key string) map[string]any { - v, _ := m[key].(map[string]any) - return v -} - func ingestDocument(ctx context.Context, tx pgx.Tx, m map[string]any) error { - slug := str(m, "slug") - title := str(m, "title") - content := str(m, "content") - entitySlug := str(m, "entity_slug") - tags := strSlice(m, "tags") + slug := seedStr(m, "slug") + title := seedStr(m, "title") + content := seedStr(m, "content") + entitySlug := seedStr(m, "entity_slug") + tags := seedStrSlice(m, "tags") atGlance, _ := m["at_glance"].(map[string]any) clRaw, _ := m["changelog"].([]any) @@ -109,7 +201,7 @@ func ingestDocument(ctx context.Context, tx pgx.Tx, m map[string]any) error { } if entitySlug != "" { - if err := createEdge(ctx, tx, entityDocSlug, entitySlug, "documents", nil); err != nil { + if err := createSeedEdge(ctx, tx, entityDocSlug, entitySlug, "documents", nil); err != nil { return fmt.Errorf("link document: %w", err) } } @@ -128,14 +220,14 @@ func ingestDocument(ctx context.Context, tx pgx.Tx, m map[string]any) error { } func ingestInvestigation(ctx context.Context, tx pgx.Tx, m map[string]any) error { - slug := str(m, "slug") - title := str(m, "title") - content := str(m, "content") - date := str(m, "date") - status := str(m, "status") - duration := str(m, "duration") - aboutSlugs := strSlice(m, "about_slugs") - tags := strSlice(m, "tags") + slug := seedStr(m, "slug") + title := seedStr(m, "title") + content := seedStr(m, "content") + date := seedStr(m, "date") + status := seedStr(m, "status") + duration := seedStr(m, "duration") + aboutSlugs := seedStrSlice(m, "about_slugs") + tags := seedStrSlice(m, "tags") entitySlug := "investigation:" + slug @@ -164,7 +256,7 @@ func ingestInvestigation(ctx context.Context, tx pgx.Tx, m map[string]any) error } for _, aboutSlug := range aboutSlugs { - if err := createEdge(ctx, tx, entitySlug, aboutSlug, "about", nil); err != nil { + if err := createSeedEdge(ctx, tx, entitySlug, aboutSlug, "about", nil); err != nil { return fmt.Errorf("link investigation about %s: %w", aboutSlug, err) } } @@ -173,12 +265,12 @@ func ingestInvestigation(ctx context.Context, tx pgx.Tx, m map[string]any) error } func ingestRunbook(ctx context.Context, tx pgx.Tx, m map[string]any) error { - slug := str(m, "slug") - name := str(m, "name") - riskClass := str(m, "risk_class") - entityType := str(m, "entity_type") - content := str(m, "content") - tags := strSlice(m, "tags") + slug := seedStr(m, "slug") + name := seedStr(m, "name") + riskClass := seedStr(m, "risk_class") + entityType := seedStr(m, "entity_type") + content := seedStr(m, "content") + tags := seedStrSlice(m, "tags") procedure, _ := m["procedure"].(map[string]any) entitySlug := "runbook:" + slug @@ -211,12 +303,12 @@ func ingestRunbook(ctx context.Context, tx pgx.Tx, m map[string]any) error { } func upsertKnowledgeEntity(ctx context.Context, tx pgx.Tx, slug, entityType, title, content, source string, tags []string) error { - id, err := getOrCreateEntity(ctx, tx, slug, entityType, title) + id, err := getOrCreateKnowledgeEntity(ctx, tx, slug, entityType, title) if err != nil { return err } - hash := contentHash(content) + hash := contentHash([]byte(content)) tagArray := toPGArray(tags) _, err = tx.Exec(ctx, @@ -229,7 +321,7 @@ func upsertKnowledgeEntity(ctx context.Context, tx pgx.Tx, slug, entityType, tit return err } -func getOrCreateEntity(ctx context.Context, tx pgx.Tx, slug, entityType, name string) (uuid.UUID, error) { +func getOrCreateKnowledgeEntity(ctx context.Context, tx pgx.Tx, slug, entityType, name 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 { @@ -253,7 +345,7 @@ func getOrCreateEntity(ctx context.Context, tx pgx.Tx, slug, entityType, name st return id, nil } -func createEdge(ctx context.Context, tx pgx.Tx, sourceSlug, targetSlug, relType string, attrs map[string]any) error { +func createSeedEdge(ctx context.Context, tx pgx.Tx, sourceSlug, targetSlug, relType string, attrs map[string]any) error { var sourceID, targetID uuid.UUID if err := tx.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", sourceSlug).Scan(&sourceID); err != nil { return fmt.Errorf("source %s: %w", sourceSlug, err) @@ -285,4 +377,4 @@ func toPGArray(tags []string) string { } out += "}" return out -} \ No newline at end of file +} diff --git a/internal/adapters/postgres/seedrepo_test.go b/internal/adapters/postgres/seedrepo_test.go new file mode 100644 index 00000000..925c11fc --- /dev/null +++ b/internal/adapters/postgres/seedrepo_test.go @@ -0,0 +1,65 @@ +package db + +import ( + "reflect" + "testing" +) + +// Tests for the seed-helper functions absorbed from internal/knowledge +// (Phase 7). + +func TestSeedStr(t *testing.T) { + cases := []struct { + name string + m map[string]any + key string + want string + }{ + {"missing key", map[string]any{}, "nope", ""}, + {"string value", map[string]any{"k": "v"}, "k", "v"}, + {"int value", map[string]any{"k": 42}, "k", ""}, + {"nil value", map[string]any{"k": nil}, "k", ""}, + {"empty string", map[string]any{"k": ""}, "k", ""}, + } + for _, c := range cases { + if got := seedStr(c.m, c.key); got != c.want { + t.Errorf("%s: seedStr(%v, %q) = %q, want %q", c.name, c.m, c.key, got, c.want) + } + } +} + +func TestSeedStrSlice(t *testing.T) { + cases := []struct { + name string + m map[string]any + key string + want []string + }{ + {"missing key", map[string]any{}, "tags", nil}, + {"string list", map[string]any{"tags": []any{"a", "b"}}, "tags", []string{"a", "b"}}, + {"mixed list drops non-strings", map[string]any{"tags": []any{"a", 1, "b"}}, "tags", []string{"a", "b"}}, + {"empty list", map[string]any{"tags": []any{}}, "tags", nil}, + } + for _, c := range cases { + if got := seedStrSlice(c.m, c.key); !reflect.DeepEqual(got, c.want) { + t.Errorf("%s: seedStrSlice(%v, %q) = %v, want %v", c.name, c.m, c.key, got, c.want) + } + } +} + +func TestToPGArray(t *testing.T) { + cases := []struct { + in []string + want string + }{ + {nil, "{}"}, + {[]string{}, "{}"}, + {[]string{"ops"}, `{"ops"}`}, + {[]string{"ops", "network"}, `{"ops","network"}`}, + } + for _, c := range cases { + if got := toPGArray(c.in); got != c.want { + t.Errorf("toPGArray(%v) = %q, want %q", c.in, got, c.want) + } + } +} diff --git a/internal/adapters/ssh/provisioner.go b/internal/adapters/ssh/provisioner.go new file mode 100644 index 00000000..3935c7ea --- /dev/null +++ b/internal/adapters/ssh/provisioner.go @@ -0,0 +1,237 @@ +package ssh + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strconv" + "strings" + + "github.com/dtoro/oikos/internal/core/ports" +) + +// Provisioner implements ports.Provisioner over CommandExecutor: the +// pct create flow absorbed from httpapi's approved-execution path +// (Phase 7). It owns the SSH-side pre-flights — template-cache +// resolution, cluster-wide VMID collision guard, gateway reachability +// on the target bridge — and the pct create itself. Guest registration +// in the entity graph is ProvisioningService's job, not this adapter's. +type Provisioner struct { + exec ports.CommandExecutor +} + +var _ ports.Provisioner = (*Provisioner)(nil) + +// NewProvisioner builds the provisioner over a command executor (the +// dial-pool executor; the target is the Proxmox host, reached directly). +func NewProvisioner(exec ports.CommandExecutor) *Provisioner { + return &Provisioner{exec: exec} +} + +func (p *Provisioner) target(in ports.LXCInput) ports.Target { + return ports.Target{Host: in.HostAddr, User: in.HostUser} +} + +// CreateLXC runs the full pct create: template pre-flight, VMID guard, +// gateway pre-flight, then create + start. Output streams to in.Sink. +func (p *Provisioner) CreateLXC(ctx context.Context, in ports.LXCInput) (ports.ProvisionResult, error) { + // Template pre-flight: resolve against what the host actually has + // cached. A hardcoded name (e.g. debian-13) fails opaquely with a raw + // `pct` error when that exact file isn't present. List the cache, then + // either validate the requested template or auto-pick the newest + // debian one; on miss, fail early with the available list so the + // operator/agent can retry with a real name. + res := p.exec.Run(ctx, p.target(in), + "ls -1 /var/lib/vz/template/cache/ 2>/dev/null | grep -E '\\.tar\\.(zst|gz|xz)$' || true", + ports.ExecOpts{}) + if res.Err != nil { + return ports.ProvisionResult{}, fmt.Errorf("list templates on %s: %w", in.HostSlug, res.Err) + } + available := []string{} + for _, l := range strings.Split(strings.TrimSpace(res.Output), "\n") { + if l = strings.TrimSpace(l); l != "" { + available = append(available, l) + } + } + in.Template = ResolveTemplate(in.Template, available) + if in.Template == "" { + return ports.ProvisionResult{}, fmt.Errorf( + "no usable LXC template on %s; available: %v", in.HostSlug, available) + } + + // VMID collision guard. Proxmox VMIDs are cluster-wide, so the model's + // guess (e.g. 132) can collide with a container on another node — pct + // create then fails with "CT N already exists on node X". Fetch the set + // of in-use VMIDs across the cluster; if the requested id is taken (or + // absent), fall back to the cluster's next free id so provisioning + // still succeeds instead of dead-ending on the operator's approval. + var vmidErr error + in.VMID, vmidErr = p.resolveVMID(ctx, in) + if vmidErr != nil { + return ports.ProvisionResult{}, vmidErr + } + + // Gateway pre-flight: for a static config, ping the gateway from the + // target HOST, on the SPECIFIC BRIDGE being requested, before spending + // 5+ minutes creating the container. Binding to the bridge + // (`ping -I `) matters: a bare `ping ` from the host can + // succeed via the host's own routing table even when the container — + // which only gets a naive on-link default route via its bridge's veth — + // can never ARP that gateway at all. Binding reproduces what the + // container will actually experience. + isStatic := in.IP != "" && !strings.EqualFold(in.IP, "dhcp") + if isStatic && in.GW != "" { + ping := p.exec.Run(ctx, p.target(in), + fmt.Sprintf("ping -I %s -c1 -W2 %s >/dev/null 2>&1 && echo PREFLIGHT_OK || echo PREFLIGHT_FAIL", in.Bridge, in.GW), + ports.ExecOpts{}) + if ping.Err != nil || !GatewayPreflightPassed(ping.Output) { + return ports.ProvisionResult{}, fmt.Errorf( + "gateway %s is not reachable from %s on bridge %s — this almost always means the bridge doesn't carry that subnet on this host (each bridge only reaches the network it's physically wired to). "+ + "Do not retry with a different gateway guess in the same subnet: find an existing LXC on this host with an IP in the same /28 and copy its exact bridge+gateway, or use DHCP instead", + in.GW, in.HostSlug, in.Bridge) + } + } + + createCmd := BuildPctCreateCmd(in) + + slog.Info("provisioner: pct create running", + "vmid", in.VMID, "hostname", in.Hostname, "cmd", createCmd) + run := p.exec.Run(ctx, p.target(in), createCmd, ports.ExecOpts{Sink: in.Sink}) + if run.Err != nil { + return ports.ProvisionResult{VMID: in.VMID, Template: in.Template, Output: run.Output}, run.Err + } + return ports.ProvisionResult{VMID: in.VMID, Template: in.Template, Output: run.Output}, nil +} + +// resolveVMID returns the VMID to use, reassigning via the cluster's +// next free id when the request is absent or already taken. +func (p *Provisioner) resolveVMID(ctx context.Context, in ports.LXCInput) (int, error) { + usedRaw := p.exec.Run(ctx, p.target(in), + `pvesh get /cluster/resources --type vm --output-format json 2>/dev/null | grep -o '"vmid":[0-9]*' | grep -o '[0-9]*' || true`, + ports.ExecOpts{}) + used := map[int]bool{} + for _, l := range strings.Fields(usedRaw.Output) { + if n, e := strconv.Atoi(strings.TrimSpace(l)); e == nil { + used[n] = true + } + } + if in.VMID != 0 && !used[in.VMID] { + return in.VMID, nil + } + if in.VMID != 0 { + slog.Info("provisioner: pct_create VMID reassigned", "requested", in.VMID) + } + nextRaw := p.exec.Run(ctx, p.target(in), `pvesh get /cluster/nextid 2>/dev/null`, ports.ExecOpts{}) + nextID, cerr := strconv.Atoi(strings.TrimSpace(nextRaw.Output)) + if nextRaw.Err != nil || cerr != nil || nextID == 0 { + return 0, fmt.Errorf("VMID %d is already in use on the cluster and could not resolve a free id", in.VMID) + } + return nextID, nil +} + +// CreateVM is reserved for the qm flow; no consumer exists yet. +func (p *Provisioner) CreateVM(_ context.Context, _ ports.VMInput) (ports.ProvisionResult, error) { + return ports.ProvisionResult{}, errors.New("provisioner: VM creation via qm not implemented yet") +} + +// BuildPctCreateCmd renders the pct create command from a +// fully-defaulted input. Pure; split out for tests. +func BuildPctCreateCmd(in ports.LXCInput) string { + privFlag := "--unprivileged 1" + if in.Privileged { + privFlag = "--unprivileged 0" + } + + features := []string{} + if in.Nesting { + features = append(features, "nesting=1") + } + if in.Privileged { + features = append(features, "keyctl=1") + } + nestingFlag := "" + if len(features) > 0 { + nestingFlag = fmt.Sprintf(" --features %s", strings.Join(features, ",")) + } + + // net0: DHCP when no static IP is given (or ip=="dhcp"). Proxmox + // rejects a gateway alongside ip=dhcp, so only add gw for a static IP. + net0 := "name=eth0,bridge=" + in.Bridge + "," + if in.IP == "" || strings.EqualFold(in.IP, "dhcp") { + net0 += "ip=dhcp" + } else { + net0 += "ip=" + in.IP + if in.GW != "" { + net0 += ",gw=" + in.GW + } + } + + templatePath := fmt.Sprintf("/var/lib/vz/template/cache/%s", in.Template) + cmd := fmt.Sprintf( + "pct create %d %s --hostname %s --cores %d --memory %d --rootfs %s:%d %s --net0 %s%s --start 1", + in.VMID, templatePath, in.Hostname, in.Cores, in.MemoryMB, + in.Storage, in.DiskGB, privFlag, net0, nestingFlag) + + if in.Nameserver != "" { + cmd += fmt.Sprintf(" --nameserver %s", in.Nameserver) + } + if in.Searchdomain != "" { + cmd += fmt.Sprintf(" --searchdomain %s", in.Searchdomain) + } + + for i, mp := range in.Mounts { + if i < 10 { // pct supports up to mp9 + cmd += fmt.Sprintf(" --mp%d %s", i, mp) + } + } + return cmd +} + +// ResolveTemplate maps a requested template name to one actually present +// in the host's template cache. Exact match wins; a bare distro hint +// (e.g. "debian-13" or "debian") matches by prefix; empty picks the +// newest debian (falling back to any) template available. Returns "" +// when nothing fits. +func ResolveTemplate(requested string, available []string) string { + if len(available) == 0 { + return "" + } + if requested != "" { + for _, a := range available { + if a == requested { + return a + } + } + for _, a := range available { + if strings.HasPrefix(a, requested) { + return a + } + } + } + // Auto-pick: prefer debian, then the lexically-greatest (newest version). + best := "" + for _, a := range available { + if strings.Contains(a, "debian") && a > best { + best = a + } + } + if best != "" { + return best + } + for _, a := range available { + if a > best { + best = a + } + } + return best +} + +// GatewayPreflightPassed interprets the PREFLIGHT_OK/PREFLIGHT_FAIL +// markers from the gateway pre-flight check. Exact-match markers — a +// prior version checked for "REACHABLE", a substring of "UNREACHABLE", +// so the check could never actually fail. Exact match plus a test make +// that bug class structurally unable to recur silently. +func GatewayPreflightPassed(out string) bool { + return strings.TrimSpace(out) == "PREFLIGHT_OK" +} diff --git a/internal/adapters/ssh/provisioner_test.go b/internal/adapters/ssh/provisioner_test.go new file mode 100644 index 00000000..a773441a --- /dev/null +++ b/internal/adapters/ssh/provisioner_test.go @@ -0,0 +1,100 @@ +package ssh + +import ( + "strings" + "testing" + + "github.com/dtoro/oikos/internal/core/ports" +) + +// TestResolveTemplate — moved from httpapi/pct_create_test.go with the +// pct flow it belongs to. +func TestResolveTemplate(t *testing.T) { + avail := []string{ + "debian-12-standard_12.7-1_amd64.tar.zst", + "debian-13-standard_13.0-1_amd64.tar.zst", + "ubuntu-24.04-standard_24.04-2_amd64.tar.zst", + } + cases := []struct { + requested string + want string + }{ + {"debian-13-standard_13.0-1_amd64.tar.zst", "debian-13-standard_13.0-1_amd64.tar.zst"}, // exact + {"debian-13", "debian-13-standard_13.0-1_amd64.tar.zst"}, // prefix + {"", "debian-13-standard_13.0-1_amd64.tar.zst"}, // auto newest debian + {"debian-99", "debian-13-standard_13.0-1_amd64.tar.zst"}, // miss prefix → auto debian + {"ubuntu-24", "ubuntu-24.04-standard_24.04-2_amd64.tar.zst"}, + {"alpine", "debian-13-standard_13.0-1_amd64.tar.zst"}, // miss → auto debian + } + for _, c := range cases { + if got := ResolveTemplate(c.requested, avail); got != c.want { + t.Errorf("ResolveTemplate(%q) = %q, want %q", c.requested, got, c.want) + } + } + if got := ResolveTemplate("debian", nil); got != "" { + t.Errorf("ResolveTemplate with no cache = %q, want empty", got) + } +} + +func TestGatewayPreflightPassed(t *testing.T) { + cases := []struct { + in string + want bool + }{ + {"PREFLIGHT_OK", true}, + {"PREFLIGHT_OK\n", true}, + {" PREFLIGHT_OK ", true}, + {"PREFLIGHT_FAIL", false}, + {"UNREACHABLE", false}, // the original substring bug + {"", false}, + } + for _, c := range cases { + if got := GatewayPreflightPassed(c.in); got != c.want { + t.Errorf("GatewayPreflightPassed(%q) = %v, want %v", c.in, got, c.want) + } + } +} + +func TestBuildPctCreateCmd(t *testing.T) { + base := ports.LXCInput{ + VMID: 137, Hostname: "grafana", Cores: 2, MemoryMB: 1024, DiskGB: 12, + IP: "192.168.8.55", GW: "192.168.8.2", Bridge: "vmbr1", + Storage: "local-lvm", Template: "debian-13-standard_13.0-1_amd64.tar.zst", + Nameserver: "192.168.8.2", Searchdomain: "hubris.network", + } + got := BuildPctCreateCmd(base) + for _, want := range []string{ + "pct create 137 /var/lib/vz/template/cache/debian-13-standard_13.0-1_amd64.tar.zst", + "--hostname grafana", + "--cores 2", + "--memory 1024", + "--rootfs local-lvm:12", + "--unprivileged 1", + "--net0 name=eth0,bridge=vmbr1,ip=192.168.8.55,gw=192.168.8.2", + "--start 1", + "--nameserver 192.168.8.2", + "--searchdomain hubris.network", + } { + if !strings.Contains(got, want) { + t.Errorf("cmd missing %q:\n%s", want, got) + } + } + if strings.Contains(got, "--features") { + t.Errorf("unprivileged non-nested should have no features flag:\n%s", got) + } + + nested := base + nested.Nesting = true + nested.Privileged = true + nested.IP = "dhcp" + got = BuildPctCreateCmd(nested) + if !strings.Contains(got, "--features nesting=1,keyctl=1") { + t.Errorf("privileged+nesting features missing:\n%s", got) + } + if !strings.Contains(got, "--unprivileged 0") { + t.Errorf("privileged should pass --unprivileged 0:\n%s", got) + } + if !strings.Contains(got, "ip=dhcp") || strings.Contains(got, "gw=") { + t.Errorf("dhcp must omit gateway:\n%s", got) + } +} diff --git a/internal/core/app/provisioning.go b/internal/core/app/provisioning.go new file mode 100644 index 00000000..799f700e --- /dev/null +++ b/internal/core/app/provisioning.go @@ -0,0 +1,239 @@ +package app + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/dtoro/oikos/internal/core/domain" + "github.com/dtoro/oikos/internal/core/ports" + "github.com/google/uuid" +) + +// ProvisioningService owns the pct_create use-case: validate + default +// the spec, run the provisioner (SSH-side template resolution, VMID +// guard, preflight, pct create), then register the guest in the entity +// graph. Absorbed the pct_create arm of httpapi's approved-execution +// path (Phase 7); the execution bookkeeping (status rows, events, log +// streaming to the SPA) stays with the calling adapter. +// +// Registration semantics are deliberately best-effort, as before: a +// failed entity/edge write is logged, never fatal — the container exists +// on the host, and the graph can be repaired by re-seeding or manual +// upsert. Failing the execution after a successful create would report +// a provisioning failure for a container that actually exists. +type ProvisioningService struct { + provisioner ports.Provisioner + resolver ports.TargetResolver + entities ports.EntityRepository + rels ports.RelationshipRepository +} + +// NewProvisioningService wires the service. +func NewProvisioningService( + provisioner ports.Provisioner, + resolver ports.TargetResolver, + entities ports.EntityRepository, + rels ports.RelationshipRepository, +) *ProvisioningService { + return &ProvisioningService{provisioner: provisioner, resolver: resolver, entities: entities, rels: rels} +} + +// CreateLXCCmd is one LXC create request. Only Hostname is required; +// zero fields get lab defaults via DefaultLXCSpec. +type CreateLXCCmd struct { + HostSlug string // the Proxmox host (host:) that runs pct + Hostname string + + VMID int + Cores int + MemoryMB int + DiskGB int + IP string + GW string + Bridge string + Storage string + Template string + Privileged bool + Nesting bool + Mounts []string + Nameserver string + Searchdomain string + + // Sink, when non-nil, receives the create command's output chunks as + // they arrive (execution-log streaming). + Sink func(stream string, chunk []byte) +} + +// DefaultLXCSpec fills the lab defaults. Kept as data (not baked into +// the adapter) so tests and future callers can see and share them. +func DefaultLXCSpec(cmd CreateLXCCmd) ports.LXCInput { + in := ports.LXCInput{ + HostSlug: cmd.HostSlug, + Hostname: cmd.Hostname, + VMID: cmd.VMID, + Cores: cmd.Cores, + MemoryMB: cmd.MemoryMB, + DiskGB: cmd.DiskGB, + IP: cmd.IP, + GW: cmd.GW, + Bridge: cmd.Bridge, + Storage: cmd.Storage, + Template: cmd.Template, + Privileged: cmd.Privileged, + Nesting: cmd.Nesting, + Mounts: cmd.Mounts, + Nameserver: cmd.Nameserver, + Searchdomain: cmd.Searchdomain, + Sink: cmd.Sink, + } + if in.Cores == 0 { + in.Cores = 1 + } + if in.MemoryMB == 0 { + in.MemoryMB = 512 + } + if in.DiskGB == 0 { + in.DiskGB = 8 + } + if in.Storage == "" { + in.Storage = "local-lvm" + } + if in.GW == "" { + in.GW = "192.168.8.2" + } + if in.Bridge == "" { + in.Bridge = "vmbr0" + } + if in.Nameserver == "" { + in.Nameserver = "192.168.8.2" + } + if in.Searchdomain == "" { + in.Searchdomain = "hubris.network" + } + return in +} + +// LXCOutcome reports a completed create: the graph slug, the VMID the +// cluster actually assigned (may differ from the request), and the +// command output. +type LXCOutcome struct { + Slug string + VMID int + Template string + Output string +} + +// CreateLXC validates the request, provisions the container on the +// resolved Proxmox host, and registers it in the graph. pct_create is +// atomic by design: create + start + register, nothing else — package +// installs and post-install scripts are the agent's own follow-up `run` +// calls so each step is individually observable and recoverable. +func (s *ProvisioningService) CreateLXC(ctx context.Context, cmd CreateLXCCmd) (LXCOutcome, error) { + if cmd.Hostname == "" { + return LXCOutcome{}, errors.New("pct_create: hostname is required") + } + + in := DefaultLXCSpec(cmd) + + addr, user, err := s.resolver.ResolveHost(ctx, in.HostSlug, "") + if err != nil { + return LXCOutcome{}, err + } + in.HostAddr, in.HostUser = addr, user + + res, err := s.provisioner.CreateLXC(ctx, in) + if err != nil { + return LXCOutcome{}, err + } + + out := LXCOutcome{ + Slug: "lxc:" + in.Hostname, + VMID: res.VMID, + Template: res.Template, + Output: res.Output, + } + s.registerLXC(ctx, in, out) + return out, nil +} + +// registerLXC best-effort writes the guest into the graph: entity +// (provisioning state, enrolled now), hosts edge from the Proxmox host, +// and the entity_status row (via the create's derived-checks path). +// Failures are logged, not returned — see the service comment. +func (s *ProvisioningService) registerLXC(ctx context.Context, in ports.LXCInput, out LXCOutcome) { + id, err := uuid.NewV7() + if err != nil { + slog.Error("provisioning: generate entity id", "error", err, "slug", out.Slug) + return + } + + now := time.Now() + attrs := map[string]any{ + "pve_id": fmt.Sprintf("%d", out.VMID), + "host": strings.TrimPrefix(in.HostSlug, "host:"), + } + if in.IP != "" && !strings.EqualFold(in.IP, "dhcp") { + attrs["ip"] = in.IP + } + + guestID := domain.UUID(id.String()) + created, err := s.entities.Create(ctx, ports.EntityCreateInput{ + Entity: domain.Entity{ + ID: guestID, + Slug: out.Slug, + Type: "lxc", + Name: in.Hostname, + State: "provisioning", + Attributes: attrs, + }, + EnrolledAt: &now, + }) + switch { + case errors.Is(err, domain.ErrAlreadyExists): + // Re-provision of a known slug: link the edge to the existing row. + existing, lerr := s.entities.BySlug(ctx, out.Slug) + if lerr != nil { + slog.Error("provisioning: resolve existing lxc entity", "error", lerr, "slug", out.Slug) + return + } + guestID = existing.ID + case err != nil: + slog.Error("provisioning: lxc entity register", "error", err, "slug", out.Slug) + return + default: + guestID = created.ID + } + + host, herr := s.entities.BySlug(ctx, in.HostSlug) + if herr != nil { + slog.Error("provisioning: resolve host entity for edge", "error", herr, "host", in.HostSlug) + return + } + if _, rerr := s.rels.Create(ctx, ports.RelationshipCreateInput{ + Relationship: domain.Relationship{ + SourceID: host.ID, + TargetID: guestID, + Type: "hosts", + Attributes: map[string]any{"provisioned_by": "nomos"}, + }, + }); rerr != nil && !errors.Is(rerr, domain.ErrAlreadyExists) { + slog.Error("provisioning: hosts edge register", "error", rerr, "host", in.HostSlug, "lxc", out.Slug) + } + + slog.Info("provisioning: lxc entity registered", "slug", out.Slug, "vmid", out.VMID, "host", in.HostSlug) +} + +// CreateVM provisions a VM via qm on a Proxmox host. Reserved for the +// qm flow; the ssh adapter reports not-implemented until it lands. +func (s *ProvisioningService) CreateVM(ctx context.Context, cmd ports.VMInput) (ports.ProvisionResult, error) { + addr, user, err := s.resolver.ResolveHost(ctx, cmd.HostSlug, "") + if err != nil { + return ports.ProvisionResult{}, err + } + cmd.HostAddr, cmd.HostUser = addr, user + return s.provisioner.CreateVM(ctx, cmd) +} diff --git a/internal/core/app/provisioning_test.go b/internal/core/app/provisioning_test.go new file mode 100644 index 00000000..1f52fca9 --- /dev/null +++ b/internal/core/app/provisioning_test.go @@ -0,0 +1,125 @@ +package app + +import ( + "context" + "errors" + "testing" + + "github.com/dtoro/oikos/internal/core/domain" + "github.com/dtoro/oikos/internal/core/ports" + "github.com/dtoro/oikos/internal/core/ports/portstest" +) + +func newProvisioningTestDeps() (*portstest.FakeProvisioner, *portstest.FakeResolver, *portstest.EntityRepo, *portstest.RelRepo, *ProvisioningService) { + prov := &portstest.FakeProvisioner{LXCRes: ports.ProvisionResult{VMID: 142, Template: "debian-13-standard_13.0-1_amd64.tar.zst", Output: "create ok"}} + resolver := &portstest.FakeResolver{Addr: "10.0.0.5", User: "root"} + entities := portstest.NewEntityRepo() + rels := portstest.NewRelRepo() + entities.Create(context.Background(), ports.EntityCreateInput{ + Entity: domain.Entity{ID: "host-1", Slug: "host:hubris", Type: "proxmox-host", Name: "hubris", State: "active"}, + }) + svc := NewProvisioningService(prov, resolver, entities, rels) + return prov, resolver, entities, rels, svc +} + +func TestProvisioningCreateLXCRequiresHostname(t *testing.T) { + _, _, _, _, svc := newProvisioningTestDeps() + _, err := svc.CreateLXC(context.Background(), CreateLXCCmd{HostSlug: "host:hubris"}) + if err == nil || err.Error() != "pct_create: hostname is required" { + t.Fatalf("got %v, want hostname-required error", err) + } +} + +func TestProvisioningCreateLXCHappyPath(t *testing.T) { + prov, resolver, entities, rels, svc := newProvisioningTestDeps() + + out, err := svc.CreateLXC(context.Background(), CreateLXCCmd{ + HostSlug: "host:hubris", + Hostname: "grafana", + IP: "192.168.8.55", + }) + if err != nil { + t.Fatalf("CreateLXC: %v", err) + } + + if out.Slug != "lxc:grafana" || out.VMID != 142 { + t.Errorf("outcome = %+v, want slug lxc:grafana vmid 142", out) + } + + // Provisioner saw the resolved endpoint and the defaults. + if len(prov.LXCs) != 1 { + t.Fatalf("provisioner calls = %d, want 1", len(prov.LXCs)) + } + in := prov.LXCs[0] + if in.HostAddr != "10.0.0.5" || in.HostUser != "root" { + t.Errorf("resolved endpoint = %s@%s, want root@10.0.0.5", in.HostUser, in.HostAddr) + } + if in.Cores != 1 || in.MemoryMB != 512 || in.DiskGB != 8 || in.Storage != "local-lvm" { + t.Errorf("defaults not applied: %+v", in) + } + if in.Bridge != "vmbr0" || in.GW != "192.168.8.2" || in.Searchdomain != "hubris.network" { + t.Errorf("network defaults not applied: %+v", in) + } + + // Entity registered: provisioning state, enrolled, vmid/host attrs. + got, ok := entities.FindBySlug("lxc:grafana") + if !ok { + t.Fatal("lxc:grafana not registered") + } + if got.Type != "lxc" || got.State != "provisioning" { + t.Errorf("entity = %s/%s, want lxc/provisioning", got.Type, got.State) + } + if got.Attributes["pve_id"] != "142" || got.Attributes["host"] != "hubris" || got.Attributes["ip"] != "192.168.8.55" { + t.Errorf("attrs = %v", got.Attributes) + } + if _, enrolled := entities.Enrolled[got.ID]; !enrolled { + t.Error("entity not enrolled") + } + + // Hosts edge from the Proxmox host to the guest. + edges, err := rels.ListFor(context.Background(), domain.UUID("host-1"), "outbound") + if err != nil || len(edges) != 1 { + t.Fatalf("host edges = %v err=%v, want 1", edges, err) + } + if edges[0].Type != "hosts" || edges[0].TargetID != got.ID { + t.Errorf("edge = %+v, want hosts → %s", edges[0], got.ID) + } + if edges[0].Attributes["provisioned_by"] != "nomos" { + t.Errorf("edge attrs = %v, want provisioned_by=nomos", edges[0].Attributes) + } + + _ = resolver // endpoint asserted via provisioner input +} + +func TestProvisioningCreateLXCProvisionerFailureRegistersNothing(t *testing.T) { + prov, _, entities, rels, svc := newProvisioningTestDeps() + prov.LXCErr = errors.New("no usable LXC template") + + if _, err := svc.CreateLXC(context.Background(), CreateLXCCmd{HostSlug: "host:hubris", Hostname: "x"}); err == nil { + t.Fatal("expected provisioner error to propagate") + } + if _, ok := entities.FindBySlug("lxc:x"); ok { + t.Error("failed create must not register an entity") + } + if edges, _ := rels.ListFor(context.Background(), domain.UUID("host-1"), "outbound"); len(edges) != 0 { + t.Errorf("failed create must not create edges, got %v", edges) + } +} + +func TestProvisioningCreateLXCRegistrationBestEffort(t *testing.T) { + _, _, entities, rels, svc := newProvisioningTestDeps() + entities.ErrStub = errors.New("db down") + + // The container exists on the host; a graph write failure must not + // fail the provisioning outcome. + out, err := svc.CreateLXC(context.Background(), CreateLXCCmd{HostSlug: "host:hubris", Hostname: "y"}) + if err != nil { + t.Fatalf("CreateLXC should survive registration failure: %v", err) + } + if out.VMID != 142 { + t.Errorf("outcome = %+v, want provisioner result", out) + } + if edges, _ := rels.ListFor(context.Background(), domain.UUID("host-1"), "outbound"); len(edges) != 0 { + t.Errorf("no edges expected when entity write failed, got %v", edges) + } +} diff --git a/internal/core/app/secrets.go b/internal/core/app/secrets.go new file mode 100644 index 00000000..50e0d96a --- /dev/null +++ b/internal/core/app/secrets.go @@ -0,0 +1,42 @@ +package app + +import ( + "context" + + "github.com/dtoro/oikos/internal/core/ports" +) + +// SecretsService is the secrets use-case surface. get/list are direct +// reads; Set is a direct write for the CLI/operator path — routing agent +// writes through the approval flow (plan §3.4) lands with the governance +// slice (PolicyService/ExecutionService), at which point the MCP +// set_secret tool converges here too. +type SecretsService struct { + backend ports.Secrets +} + +// NewSecretsService wires the service over a secrets backend +// (Infisical primary, SOPS DR fallback). +func NewSecretsService(backend ports.Secrets) *SecretsService { + return &SecretsService{backend: backend} +} + +// Get retrieves one secret value by key. +func (s *SecretsService) Get(ctx context.Context, key string) (string, error) { + return s.backend.Get(ctx, key) +} + +// List returns all secret keys (no values). +func (s *SecretsService) List(ctx context.Context) ([]string, error) { + return s.backend.List(ctx) +} + +// Set stores or updates a secret. +func (s *SecretsService) Set(ctx context.Context, key, value string) error { + return s.backend.Set(ctx, key, value) +} + +// Name reports the active backend (for diagnostics). +func (s *SecretsService) Name() string { + return s.backend.Name() +} diff --git a/internal/core/app/seed.go b/internal/core/app/seed.go new file mode 100644 index 00000000..6b5c1547 --- /dev/null +++ b/internal/core/app/seed.go @@ -0,0 +1,95 @@ +package app + +import ( + "context" + "fmt" + "log/slog" + "os" + + "github.com/dtoro/oikos/internal/core/ports" +) + +// SeedService is the bootstrap/DR use-case: ingest the seeds/*.yaml files +// into the DB (idempotent, ontology-validated) and regenerate them from +// the DB for version control. Absorbed the ingest orchestration that +// lived in cmd/oikos runSeed and the export entry point in runExport +// (Phase 7); the SQL stays behind ports.SeedRepository in the postgres +// adapter. +type SeedService struct { + repo ports.SeedRepository +} + +// NewSeedService wires the service. +func NewSeedService(repo ports.SeedRepository) *SeedService { + return &SeedService{repo: repo} +} + +// Ingest reads the seed YAMLs from dir and ingests them in dependency +// order: ontology (types) before inventory (entities/edges validated +// against them), then policy, then knowledge — which is optional and +// skipped silently when absent. Files whose content hash is unchanged +// are no-ops at the repository layer. +func (s *SeedService) Ingest(ctx context.Context, dir string) error { + type step struct { + file string + ingest func(ctx context.Context, filename string, content []byte) (ports.SeedCounts, bool, error) + optional bool + } + steps := []step{ + {"ontology.yaml", s.repo.IngestOntology, false}, + {"inventory.yaml", s.repo.IngestInventory, false}, + {"policy.yaml", s.repo.IngestPolicy, false}, + {"knowledge.yaml", s.repo.IngestKnowledge, true}, + } + + for _, st := range steps { + content, err := os.ReadFile(dir + "/" + st.file) + if err != nil { + if os.IsNotExist(err) && st.optional { + slog.Info("knowledge seed not found, skipping") + continue + } + return fmt.Errorf("read %s seed: %w", st.file, err) + } + + counts, applied, err := st.ingest(ctx, st.file, content) + if err != nil { + return err + } + if !applied { + continue + } + + switch st.file { + case "ontology.yaml": + slog.Info("ontology ingested", + "lifecycles", counts.Lifecycles, + "entity_types", counts.EntityTypes, + "relationship_types", counts.RelationshipTypes) + case "inventory.yaml": + slog.Info("inventory ingested", + "entities", counts.Entities, + "relationships", counts.Relationships) + case "policy.yaml": + slog.Info("policy ingested", + "risk_classes", counts.RiskClasses, + "approval_rules", counts.ApprovalRules, + "autonomy_settings", counts.AutonomySettings) + case "knowledge.yaml": + slog.Info("knowledge ingested", + "documents", counts.Documents, + "investigations", counts.Investigations, + "runbooks", counts.Runbooks) + } + } + + slog.Info("seed ingest complete") + return nil +} + +// Export regenerates the seed YAMLs from the DB (DR / version control). +// Deterministic: the repository orders lists and sorts map keys so +// export → ingest → export is byte-stable. +func (s *SeedService) Export(ctx context.Context) (map[string][]byte, error) { + return s.repo.Export(ctx) +} diff --git a/internal/core/app/seed_test.go b/internal/core/app/seed_test.go new file mode 100644 index 00000000..0cf8f459 --- /dev/null +++ b/internal/core/app/seed_test.go @@ -0,0 +1,98 @@ +package app + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + + "github.com/dtoro/oikos/internal/core/ports/portstest" +) + +func writeSeedFile(t *testing.T, dir, name, content string) { + t.Helper() + if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0644); err != nil { + t.Fatal(err) + } +} + +func TestSeedServiceIngestOrder(t *testing.T) { + dir := t.TempDir() + writeSeedFile(t, dir, "ontology.yaml", "a: 1") + writeSeedFile(t, dir, "inventory.yaml", "b: 2") + writeSeedFile(t, dir, "policy.yaml", "c: 3") + writeSeedFile(t, dir, "knowledge.yaml", "d: 4") + + repo := portstest.NewSeedRepo() + if err := NewSeedService(repo).Ingest(context.Background(), dir); err != nil { + t.Fatalf("Ingest: %v", err) + } + + want := []string{"ontology.yaml", "inventory.yaml", "policy.yaml", "knowledge.yaml"} + if len(repo.Order) != len(want) { + t.Fatalf("ingested %v, want %v", repo.Order, want) + } + for i, f := range want { + if repo.Order[i] != f { + t.Errorf("order[%d] = %s, want %s", i, repo.Order[i], f) + } + } +} + +func TestSeedServiceKnowledgeOptional(t *testing.T) { + dir := t.TempDir() + writeSeedFile(t, dir, "ontology.yaml", "a: 1") + writeSeedFile(t, dir, "inventory.yaml", "b: 2") + writeSeedFile(t, dir, "policy.yaml", "c: 3") + + repo := portstest.NewSeedRepo() + if err := NewSeedService(repo).Ingest(context.Background(), dir); err != nil { + t.Fatalf("Ingest without knowledge seed: %v", err) + } + for _, f := range repo.Order { + if f == "knowledge.yaml" { + t.Error("knowledge.yaml must be skipped when absent") + } + } +} + +func TestSeedServiceMissingStructuralSeedFails(t *testing.T) { + dir := t.TempDir() + writeSeedFile(t, dir, "ontology.yaml", "a: 1") + // inventory.yaml and policy.yaml absent — both are required. + + repo := portstest.NewSeedRepo() + err := NewSeedService(repo).Ingest(context.Background(), dir) + if err == nil { + t.Fatal("missing inventory seed must fail the ingest") + } + if repo.Order[len(repo.Order)-1] != "ontology.yaml" { + t.Errorf("ingest stopped at %v, want ontology.yaml only", repo.Order) + } +} + +func TestSeedServiceIngestErrorPropagates(t *testing.T) { + dir := t.TempDir() + writeSeedFile(t, dir, "ontology.yaml", "a: 1") + writeSeedFile(t, dir, "inventory.yaml", "b: 2") + + repo := portstest.NewSeedRepo() + repo.IngestErr["inventory.yaml"] = errors.New("entity bad: unknown type") + err := NewSeedService(repo).Ingest(context.Background(), dir) + if err == nil || !errors.Is(err, repo.IngestErr["inventory.yaml"]) { + t.Fatalf("got %v, want ingest error", err) + } +} + +func TestSeedServiceExport(t *testing.T) { + repo := portstest.NewSeedRepo() + repo.ExportFiles = map[string][]byte{"ontology.yaml": []byte("x")} + files, err := NewSeedService(repo).Export(context.Background()) + if err != nil { + t.Fatalf("Export: %v", err) + } + if string(files["ontology.yaml"]) != "x" { + t.Errorf("export = %v", files) + } +} diff --git a/internal/core/ports/entities.go b/internal/core/ports/entities.go index 26e888e8..04ed687a 100644 --- a/internal/core/ports/entities.go +++ b/internal/core/ports/entities.go @@ -60,6 +60,9 @@ type EntityCreateInput struct { Audit []AuditEntry Event *Event Idempotency *Idempotency + // EnrolledAt, when set, stamps the entities.enrolled_at column (the + // client-enrollment / provisioning marker). Zero for ordinary creates. + EnrolledAt *time.Time } // EntityUpdateInput mutates an entity atomically. ExpectedVersion is the diff --git a/internal/core/ports/execution.go b/internal/core/ports/execution.go index a6245d1a..9b3156fc 100644 --- a/internal/core/ports/execution.go +++ b/internal/core/ports/execution.go @@ -47,28 +47,66 @@ type TargetResolver interface { IsGuest(entityType string) bool } -// Provisioner creates guests via pct/qm on a Proxmox host (Phase 7 fills -// the input payloads in; signatures firm up with ProvisioningService). +// Provisioner creates guests via pct/qm on a Proxmox host. The ssh +// adapter implements it over CommandExecutor: template-cache resolution, +// cluster-wide VMID collision guard, gateway preflight, and the pct/qm +// create itself. DB registration of the created guest is NOT part of the +// port — ProvisioningService owns that over the entity/relationship +// repositories. type Provisioner interface { - CreateLXC(ctx context.Context, host domain.UUID, input LXCInput) (domain.UUID, error) - CreateVM(ctx context.Context, host domain.UUID, input VMInput) (domain.UUID, error) + CreateLXC(ctx context.Context, input LXCInput) (ProvisionResult, error) + CreateVM(ctx context.Context, input VMInput) (ProvisionResult, error) } -// LXCInput is a placeholder until ProvisioningService (Phase 7) fixes the -// create payloads; declared now so the port surface is complete. +// LXCInput is a fully-defaulted LXC create spec (ProvisioningService +// applies defaults before calling). HostAddr/HostUser are the resolved +// SSH endpoint of the Proxmox host; Sink, when non-nil, receives the +// create command's output chunks as they arrive. type LXCInput struct { - Name string - Template string - Cores int - MemoryMB int - DiskGB int + HostAddr string + HostUser string + HostSlug string + + Hostname string + VMID int + Cores int + MemoryMB int + DiskGB int + IP string + GW string + Bridge string + Storage string + Template string + Privileged bool + Nesting bool + Mounts []string + Nameserver string + Searchdomain string + + Sink func(stream string, chunk []byte) } // VMInput mirrors LXCInput for VM creation via qm. type VMInput struct { + HostAddr string + HostUser string + HostSlug string + Name string + VMID int TemplateID int Cores int MemoryMB int DiskGB int + + Sink func(stream string, chunk []byte) +} + +// ProvisionResult reports what the provisioner actually did — the VMID +// may differ from the request (cluster collision guard reassigns), and +// the template is the concrete cache entry picked. +type ProvisionResult struct { + VMID int + Template string + Output string } diff --git a/internal/core/ports/portstest/fakes.go b/internal/core/ports/portstest/fakes.go index 7ca60660..2cad134c 100644 --- a/internal/core/ports/portstest/fakes.go +++ b/internal/core/ports/portstest/fakes.go @@ -8,6 +8,7 @@ import ( "fmt" "strings" "sync" + "time" "github.com/dtoro/oikos/internal/core/domain" "github.com/dtoro/oikos/internal/core/ports" @@ -26,6 +27,7 @@ type EntityRepo struct { Checks map[domain.UUID][]ports.CheckDef Idempotent map[string]ports.IdempotentResponse Rederived []domain.UUID + Enrolled map[domain.UUID]time.Time ErrStub error // returned by every command when set } @@ -115,11 +117,17 @@ func (r *EntityRepo) Create(_ context.Context, in ports.EntityCreateInput) (doma in.Entity.ID = domain.UUID(fmt.Sprintf("fake-entity-%03d", r.nextID)) } if _, dup := r.bySlug[in.Entity.Slug]; dup { - return domain.Entity{}, domain.ErrConflict + return domain.Entity{}, domain.ErrAlreadyExists } if in.Entity.Version == 0 { in.Entity.Version = 1 } + if in.EnrolledAt != nil { + if r.Enrolled == nil { + r.Enrolled = make(map[domain.UUID]time.Time) + } + r.Enrolled[in.Entity.ID] = *in.EnrolledAt + } r.store(in.Entity) r.Audits = append(r.Audits, in.Audit...) if in.Event != nil { @@ -288,3 +296,192 @@ func (p *SpyPublisher) Publish(_ context.Context, event ports.Event) error { p.Events = append(p.Events, event) return p.Err } + +// RelRepo is an in-memory ports.RelationshipRepository. +type RelRepo struct { + mu sync.Mutex + edges []domain.Relationship + // ErrStubFor maps a target entity ID to an error to return when an + // edge onto that target is created (for best-effort-path tests). + ErrStubFor map[domain.UUID]error +} + +// NewRelRepo builds an empty in-memory relationship repository. +func NewRelRepo() *RelRepo { return &RelRepo{} } + +// Create stores the edge as-is. +func (r *RelRepo) Create(_ context.Context, in ports.RelationshipCreateInput) (domain.Relationship, error) { + r.mu.Lock() + defer r.mu.Unlock() + if r.ErrStubFor != nil { + if err, ok := r.ErrStubFor[in.Relationship.TargetID]; ok { + return domain.Relationship{}, err + } + } + r.edges = append(r.edges, in.Relationship) + return in.Relationship, nil +} + +// End soft-deletes matching current edges. +func (r *RelRepo) End(_ context.Context, source, target domain.UUID, relType string) error { + r.mu.Lock() + defer r.mu.Unlock() + kept := r.edges[:0] + for _, e := range r.edges { + if e.SourceID == source && e.TargetID == target && e.Type == relType { + continue + } + kept = append(kept, e) + } + r.edges = kept + return nil +} + +// ListFor returns edges touching the entity in the given direction +// ("outbound", "inbound", or both). +func (r *RelRepo) ListFor(_ context.Context, entityID domain.UUID, direction string) ([]domain.Relationship, error) { + r.mu.Lock() + defer r.mu.Unlock() + var out []domain.Relationship + for _, e := range r.edges { + switch direction { + case "outbound": + if e.SourceID == entityID { + out = append(out, e) + } + case "inbound": + if e.TargetID == entityID { + out = append(out, e) + } + default: + if e.SourceID == entityID || e.TargetID == entityID { + out = append(out, e) + } + } + } + return out, nil +} + +// FakeResolver resolves every slug onto a fixed endpoint; check-shaped +// lookups can be overridden per type. +type FakeResolver struct { + Addr string + User string + Err error + // ErrForType maps entity types to a resolve error (check path). + ErrForType map[string]error +} + +// ResolveExecTarget resolves any slug to the fixed endpoint. +func (r *FakeResolver) ResolveExecTarget(_ context.Context, _ string) (ports.Target, error) { + if r.Err != nil { + return ports.Target{}, r.Err + } + return ports.Target{Host: r.Addr, User: r.User}, nil +} + +// ResolveForCheck resolves a check target by entity type. +func (r *FakeResolver) ResolveForCheck(_ context.Context, _ domain.UUID, entityType string) (ports.Target, error) { + if err, ok := r.ErrForType[entityType]; ok { + return ports.Target{}, err + } + return r.ResolveExecTarget(context.Background(), entityType) +} + +// ResolveHost resolves any host slug to the fixed endpoint. +func (r *FakeResolver) ResolveHost(_ context.Context, _, _ string) (string, string, error) { + if r.Err != nil { + return "", "", r.Err + } + return r.Addr, r.User, nil +} + +// IsGuest marks guest types reached via pct/qm. +func (r *FakeResolver) IsGuest(entityType string) bool { + return entityType == "lxc" || entityType == "vm" +} + +// FakeProvisioner records provision requests and replies with canned +// results (default: a successful create echoing the request's VMID). +type FakeProvisioner struct { + mu sync.Mutex + LXCs []ports.LXCInput + VMs []ports.VMInput + LXCRes ports.ProvisionResult + LXCErr error + VMErr error +} + +// CreateLXC records the request and replies with the canned result. +func (p *FakeProvisioner) CreateLXC(_ context.Context, in ports.LXCInput) (ports.ProvisionResult, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.LXCs = append(p.LXCs, in) + return p.LXCRes, p.LXCErr +} + +// CreateVM records the request and replies with the canned error. +func (p *FakeProvisioner) CreateVM(_ context.Context, in ports.VMInput) (ports.ProvisionResult, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.VMs = append(p.VMs, in) + return ports.ProvisionResult{}, p.VMErr +} + +// SeedRepo is an in-memory ports.SeedRepository. Files map to canned +// (counts, applied) pairs; every ingest records the file content. +type SeedRepo struct { + mu sync.Mutex + Files map[string][]byte + Applied map[string]bool + IngestErr map[string]error + // Order records the files ingested, in order. + Order []string + ExportFiles map[string][]byte + ExportErr error +} + +// NewSeedRepo builds an empty in-memory seed repository. +func NewSeedRepo() *SeedRepo { + return &SeedRepo{ + Files: make(map[string][]byte), + Applied: make(map[string]bool), + IngestErr: make(map[string]error), + } +} + +func (r *SeedRepo) ingest(file string, content []byte) (ports.SeedCounts, bool, error) { + r.mu.Lock() + defer r.mu.Unlock() + if err := r.IngestErr[file]; err != nil { + return ports.SeedCounts{}, false, err + } + r.Files[file] = content + r.Order = append(r.Order, file) + return ports.SeedCounts{}, r.Applied[file], nil +} + +// IngestOntology records the ontology seed file. +func (r *SeedRepo) IngestOntology(_ context.Context, file string, content []byte) (ports.SeedCounts, bool, error) { + return r.ingest(file, content) +} + +// IngestInventory records the inventory seed file. +func (r *SeedRepo) IngestInventory(_ context.Context, file string, content []byte) (ports.SeedCounts, bool, error) { + return r.ingest(file, content) +} + +// IngestPolicy records the policy seed file. +func (r *SeedRepo) IngestPolicy(_ context.Context, file string, content []byte) (ports.SeedCounts, bool, error) { + return r.ingest(file, content) +} + +// IngestKnowledge records the knowledge seed file. +func (r *SeedRepo) IngestKnowledge(_ context.Context, file string, content []byte) (ports.SeedCounts, bool, error) { + return r.ingest(file, content) +} + +// Export returns the canned export payload. +func (r *SeedRepo) Export(_ context.Context) (map[string][]byte, error) { + return r.ExportFiles, r.ExportErr +} diff --git a/internal/core/ports/seeds.go b/internal/core/ports/seeds.go new file mode 100644 index 00000000..f360a099 --- /dev/null +++ b/internal/core/ports/seeds.go @@ -0,0 +1,43 @@ +package ports + +import "context" + +// SeedCounts summarizes what one seed ingest wrote, by section. Sections +// the file does not exercise stay zero. +type SeedCounts struct { + Lifecycles int + EntityTypes int + RelationshipTypes int + Entities int + Relationships int + Checks int + RiskClasses int + ApprovalRules int + AutonomySettings int + Documents int + Investigations int + Runbooks int +} + +// SeedRepository is the seed aggregate: bootstrap ingest (seeds/*.yaml → +// DB) and the inverse export (DB → seeds/*.yaml for DR / version +// control). +// +// Each Ingest method is one transaction — parse, validate against the +// ontology, write, and record the file's content hash in seed_versions — +// and a no-op (applied=false) when the hash is unchanged, so re-running +// `oikos seed` is idempotent. Validation failures roll back the whole +// file. Ingest order across files (ontology before inventory) is a +// SeedService concern; within inventory, default checks derive only after +// relationships exist (a service inherits its container's address). +// +// Export regenerates the three structural seed YAMLs deterministically +// (sorted maps, ordered lists) so export → ingest → export is +// byte-stable. Runtime state (cognition layer) is excluded. +type SeedRepository interface { + IngestOntology(ctx context.Context, filename string, content []byte) (counts SeedCounts, applied bool, err error) + IngestInventory(ctx context.Context, filename string, content []byte) (counts SeedCounts, applied bool, err error) + IngestPolicy(ctx context.Context, filename string, content []byte) (counts SeedCounts, applied bool, err error) + IngestKnowledge(ctx context.Context, filename string, content []byte) (counts SeedCounts, applied bool, err error) + Export(ctx context.Context) (map[string][]byte, error) +} diff --git a/internal/httpapi/actuator.go b/internal/httpapi/actuator.go index 4ec7cbe3..d223c99b 100644 --- a/internal/httpapi/actuator.go +++ b/internal/httpapi/actuator.go @@ -7,13 +7,13 @@ import ( "fmt" "log/slog" "os" - "strconv" "strings" "time" "github.com/dtoro/oikos/internal/actuator" "github.com/dtoro/oikos/internal/adapters/postgres" "github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen" + "github.com/dtoro/oikos/internal/core/app" "github.com/dtoro/oikos/internal/execlog" "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" @@ -78,10 +78,6 @@ const sshExecTimeout = 10 * time.Minute // remote end produced it. Shared implementation lives in internal/actuator // (actuator.streamWriter / actuator.RunStreaming). -func sshExec(ctx context.Context, host, user, command string) (string, error) { - return sshExecStream(ctx, host, user, command, nil) -} - // sshExecStream runs a command and reports its combined output, forwarding // each chunk to sink as it arrives. A nil sink behaves exactly as before. func sshExecStream(ctx context.Context, host, user, command string, sink execlog.Sink) (string, error) { @@ -224,7 +220,7 @@ func closePlanStepForExecution(ctx context.Context, pool *db.Pool, execID uuid.U map[string]any{"step_id": stepID, "seq": seq, "status": stepStatus, "execution_id": execID.String()}) } -func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, targetSlug string, actionStr string) { +func (s *Server) executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, targetSlug string, actionStr string) { slog.Info("httpapi: executing approved action", "execution_id", execID, "target", targetSlug, "action", actionStr) host, user, wrap, err := resolveRunTarget(ctx, pool, targetSlug) @@ -320,234 +316,35 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": err.Error()}) return } - // Only hostname is required. vmid is optional — when 0 (or later found - // to collide) the VMID guard below assigns a free cluster id. - if cfg.Hostname == "" { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, - execID, `{"error":"pct_create: hostname is required"}`) - emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": "missing hostname"}) - return - } - if cfg.Cores == 0 { - cfg.Cores = 1 - } - if cfg.Memory == 0 { - cfg.Memory = 512 - } - if cfg.DiskGB == 0 { - cfg.DiskGB = 8 - } - if cfg.Storage == "" { - cfg.Storage = "local-lvm" - } - if cfg.GW == "" { - cfg.GW = "192.168.8.2" - } - if cfg.Nameserver == "" { - cfg.Nameserver = "192.168.8.2" - } - if cfg.Searchdomain == "" { - cfg.Searchdomain = "hubris.network" - } - // Template pre-flight: resolve against what the host actually has - // cached. A hardcoded name (e.g. debian-13) fails opaquely with a raw - // `pct` error when that exact file isn't present. List the cache, then - // either validate the requested template or auto-pick the newest - // debian one; on miss, fail early with the available list so the - // operator/agent can retry with a real name. - cacheList, tplErr := sshExec(ctx, host, user, "ls -1 /var/lib/vz/template/cache/ 2>/dev/null | grep -E '\\.tar\\.(zst|gz|xz)$' || true") - available := []string{} - for _, l := range strings.Split(strings.TrimSpace(cacheList), "\n") { - if l = strings.TrimSpace(l); l != "" { - available = append(available, l) - } - } - if tplErr != nil { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, - execID, jsonErr("list templates on %s: %s", targetSlug, tplErr.Error())) - emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": tplErr.Error()}) - return - } - cfg.Template = resolveTemplate(cfg.Template, available) - if cfg.Template == "" { - msg := fmt.Sprintf("no usable LXC template on %s. Available: %v", targetSlug, available) - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, - execID, jsonErr("%s", msg)) - emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": msg}) - return - } - - // VMID collision guard. Proxmox VMIDs are cluster-wide, so the model's - // guess (e.g. 132) can collide with a container on another node — pct - // create then fails with "CT N already exists on node X". Fetch the set - // of in-use VMIDs across the cluster; if the requested id is taken (or - // absent), fall back to the cluster's next free id so provisioning - // still succeeds instead of dead-ending on the operator's approval. - usedRaw, _ := sshExec(ctx, host, user, `pvesh get /cluster/resources --type vm --output-format json 2>/dev/null | grep -o '"vmid":[0-9]*' | grep -o '[0-9]*' || true`) - used := map[int]bool{} - for _, l := range strings.Fields(usedRaw) { - if n, e := strconv.Atoi(strings.TrimSpace(l)); e == nil { - used[n] = true - } - } - if cfg.VMID == 0 || used[cfg.VMID] { - nextRaw, nerr := sshExec(ctx, host, user, `pvesh get /cluster/nextid 2>/dev/null`) - nextID, cerr := strconv.Atoi(strings.TrimSpace(nextRaw)) - if nerr != nil || cerr != nil || nextID == 0 { - msg := fmt.Sprintf("VMID %d is already in use on the cluster and could not resolve a free id", cfg.VMID) - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, - execID, jsonErr("%s", msg)) - emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": msg}) - return - } - slog.Info("httpapi: pct_create VMID reassigned", "requested", cfg.VMID, "assigned", nextID) - cfg.VMID = nextID - } - - privFlag := "--unprivileged 1" - if cfg.Privileged { - privFlag = "--unprivileged 0" - } - - nestingFlag := "" - features := []string{} - if cfg.Nesting { - features = append(features, "nesting=1") - } - if cfg.Privileged { - features = append(features, "keyctl=1") - } - if len(features) > 0 { - nestingFlag = fmt.Sprintf(" --features %s", strings.Join(features, ",")) - } - - if cfg.Bridge == "" { - cfg.Bridge = "vmbr0" - } - - // net0: DHCP when no static IP is given (or ip=="dhcp"). Proxmox - // rejects a gateway alongside ip=dhcp, so only add gw for a static IP. - net0 := "name=eth0,bridge=" + cfg.Bridge + "," - isStatic := cfg.IP != "" && !strings.EqualFold(cfg.IP, "dhcp") - if !isStatic { - net0 += "ip=dhcp" - } else { - net0 += "ip=" + cfg.IP - if cfg.GW != "" { - net0 += ",gw=" + cfg.GW - } - } - - // Pre-flight: for a static config, ping the gateway from the target - // HOST, on the SPECIFIC BRIDGE being requested, before spending 5+ - // minutes creating the container. This is the check that would have - // caught the real TypeType failure immediately instead of after a - // full provision attempt. - // - // Binding to the bridge (`ping -I `) matters and was found - // live: a plain unqualified `ping ` from the host can succeed via - // the host's own routing table (multiple routes, possibly through an - // upstream router) even when the *container* — which only gets a - // naive on-link default route via its bridge's veth — can never ARP - // that gateway at all. Confirmed on `strong`: bare `ping 192.168.8.2` - // succeeded (via the host's default route), but a container actually - // attached to vmbr0 showed 100% packet loss trying to reach the same - // address, because vmbr0 doesn't carry that subnet's L2 segment. - // Binding to the bridge interface reproduces what the container will - // actually experience, not what the host's broader routing table can - // reach. - if isStatic && cfg.GW != "" { - pingOut, pingErr := sshExec(ctx, host, user, fmt.Sprintf("ping -I %s -c1 -W2 %s >/dev/null 2>&1 && echo PREFLIGHT_OK || echo PREFLIGHT_FAIL", cfg.Bridge, cfg.GW)) - if pingErr != nil || !gatewayPreflightPassed(pingOut) { - msg := fmt.Sprintf( - "gateway %s is not reachable from %s on bridge %s — this almost always means the bridge doesn't carry that subnet on this host (each bridge only reaches the network it's physically wired to). "+ - "Do not retry with a different gateway guess in the same subnet: find an existing LXC on this host with an IP in the same /28 and copy its exact bridge+gateway, or use DHCP instead.", - cfg.GW, targetSlug, cfg.Bridge) - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, - execID, jsonErr("%s", msg)) - emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": msg}) - return - } - } - - templatePath := fmt.Sprintf("/var/lib/vz/template/cache/%s", cfg.Template) - createCmd := fmt.Sprintf( - "pct create %d %s --hostname %s --cores %d --memory %d --rootfs %s:%d %s --net0 %s%s --start 1", - cfg.VMID, templatePath, cfg.Hostname, cfg.Cores, cfg.Memory, - cfg.Storage, cfg.DiskGB, privFlag, net0, nestingFlag) - - if cfg.Nameserver != "" { - createCmd += fmt.Sprintf(" --nameserver %s", cfg.Nameserver) - } - if cfg.Searchdomain != "" { - createCmd += fmt.Sprintf(" --searchdomain %s", cfg.Searchdomain) - } - - // Add mount points - for i, mp := range cfg.Mounts { - if i < 10 { // pct supports up to mp9 - createCmd += fmt.Sprintf(" --mp%d %s", i, mp) - } - } - - slog.Info("httpapi: pct_create running", "vmid", cfg.VMID, "hostname", cfg.Hostname, "cmd", createCmd) - output, err = sshExecStream(ctx, host, user, createCmd, sink) - - // pct_create is now DELIBERATELY ATOMIC: create + start + register, - // nothing else. It used to also run apt installs and a post_install - // script inline as one black-box multi-minute SSH call — the agent - // got back a single opaque success/fail for the whole thing with no - // way to see (or fix) which step actually broke. That's the opposite - // of what makes an agent able to recover from errors. - // - // Installing packages, running post_install, and verifying the - // service now happen as the agent's OWN follow-up `run` calls against - // the new lxc: target — each one is synchronous (in an - // active assent window) or individually gated, so the agent observes - // every step's real output and can diagnose + retry the exact thing - // that failed instead of re-doing the whole container. See SOUL.md - // "After pct_create: you drive the install" and provisionScript's - // surviving role (DNS self-heal) is now something the agent invokes - // itself via `run`, not something baked into this handler. - // - // cfg.Services/cfg.PostInstall are intentionally no longer read here. - - // On success, register the entity in the DB with proper relationships - if err == nil { - slug := "lxc:" + cfg.Hostname - var lxcID uuid.UUID - lxcID, _ = uuid.NewV7() - attrs := map[string]any{ - "pve_id": fmt.Sprintf("%d", cfg.VMID), - "host": strings.TrimPrefix(targetSlug, "host:"), - "ip": cfg.IP, - } - attrsJSON, _ := json.Marshal(attrs) - _, insErr := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, enrolled_at) - VALUES ($1, $2, 'lxc', $3, 'provisioning', $4, now()) ON CONFLICT (slug) DO NOTHING`, lxcID, slug, cfg.Hostname, attrsJSON) - if insErr != nil { - slog.Error("httpapi: pct_create entity insert", "error", insErr, "slug", slug) - } - - // Create hosts relationship: Proxmox host → LXC - var hostID uuid.UUID - if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&hostID); err == nil { - _, relErr := pool.Exec(ctx, `INSERT INTO relationships (source_id, target_id, type, attributes, valid_from) - VALUES ($1, $2, 'hosts', '{"provisioned_by":"nomos"}'::jsonb, now())`, hostID, lxcID) - if relErr != nil { - slog.Error("httpapi: pct_create relationship insert", "error", relErr, "host", targetSlug, "lxc", slug) - } - } - - // Create entity_status row for health tracking - pool.Exec(ctx, `INSERT INTO entity_status (entity_id, health, last_check_at) - VALUES ($1, 'unknown', now()) ON CONFLICT (entity_id) DO NOTHING`, lxcID) - + // The flow (spec defaults, template/VMID pre-flights, pct create, + // graph registration) lives in ProvisioningService + the ssh + // provisioner adapter since Phase 7; this handler only maps the + // wire payload and streams the create output to the execution log. + outcome, perr := s.provisioning.CreateLXC(ctx, app.CreateLXCCmd{ + HostSlug: targetSlug, + Hostname: cfg.Hostname, + VMID: cfg.VMID, + Cores: cfg.Cores, + MemoryMB: cfg.Memory, + DiskGB: cfg.DiskGB, + IP: cfg.IP, + GW: cfg.GW, + Bridge: cfg.Bridge, + Storage: cfg.Storage, + Template: cfg.Template, + Privileged: bool(cfg.Privileged), + Nesting: bool(cfg.Nesting), + Mounts: cfg.Mounts, + Nameserver: cfg.Nameserver, + Searchdomain: cfg.Searchdomain, + Sink: sink, + }) + output = outcome.Output + err = perr + if perr == nil { emitExecutionEvent(ctx, pool, execID, "executing", map[string]any{ - "lxc_slug": slug, "vmid": cfg.VMID, "host": targetSlug, + "lxc_slug": outcome.Slug, "vmid": outcome.VMID, "host": targetSlug, }) - - slog.Info("httpapi: pct_create entity registered", "slug", slug, "vmid", cfg.VMID, "host", targetSlug) } case "run": @@ -612,52 +409,3 @@ func jsonErr(format string, args ...any) []byte { b, _ := json.Marshal(map[string]any{"error": fmt.Sprintf(format, args...)}) return b } - -// resolveTemplate maps a requested template name to one actually present in -// the host's template cache. Exact match wins; a bare distro hint (e.g. -// "debian-13" or "debian") matches by prefix; empty picks the newest debian -// (falling back to any) template available. Returns "" when nothing fits. -// gatewayPreflightPassed interprets the PREFLIGHT_OK/PREFLIGHT_FAIL markers -// from the pct_create gateway pre-flight check. Pulled out as its own -// function (rather than an inline strings.Contains at the call site) so it's -// unit-testable: a prior version checked for "REACHABLE", which is a -// substring of "UNREACHABLE" — the check could never actually fail, and it -// took a live deployment to notice. Exact-match markers plus a test make -// that specific bug class structurally unable to recur silently. -func gatewayPreflightPassed(out string) bool { - return strings.TrimSpace(out) == "PREFLIGHT_OK" -} - -func resolveTemplate(requested string, available []string) string { - if len(available) == 0 { - return "" - } - if requested != "" { - for _, a := range available { - if a == requested { - return a - } - } - for _, a := range available { - if strings.HasPrefix(a, requested) { - return a - } - } - } - // Auto-pick: prefer debian, then the lexically-greatest (newest version). - best := "" - for _, a := range available { - if strings.Contains(a, "debian") && a > best { - best = a - } - } - if best != "" { - return best - } - for _, a := range available { - if a > best { - best = a - } - } - return best -} diff --git a/internal/httpapi/api_test.go b/internal/httpapi/api_test.go index 803f2f5a..ceab9f08 100644 --- a/internal/httpapi/api_test.go +++ b/internal/httpapi/api_test.go @@ -19,6 +19,7 @@ import ( "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/adapters/postgres" "github.com/dtoro/oikos/internal/core/app" + "github.com/dtoro/oikos/internal/core/ports/portstest" "github.com/jackc/pgx/v5" ) @@ -97,7 +98,14 @@ func newTestHandler(t *testing.T, cfg config.Config) http.Handler { repo := db.NewEntityRepo(pool) onto := db.NewOntologyRepo(pool, time.Minute) - return NewHandler(handlerCtx, pool, cfg, app.NewEntityService(repo, onto), repo, db.NewEntityReader(pool), app.NewRelationshipService(db.NewRelRepo(pool), onto)) + // Provisioning/seeds get real services over fakes/real repos: tests + // below don't exercise pct_create, but the export endpoint does hit + // SeedService, and a nil would panic. + provisioning := app.NewProvisioningService( + &portstest.FakeProvisioner{}, &portstest.FakeResolver{}, + repo, db.NewRelRepo(pool)) + seeds := app.NewSeedService(db.NewSeedRepo(pool)) + return NewHandler(handlerCtx, pool, cfg, app.NewEntityService(repo, onto), repo, db.NewEntityReader(pool), app.NewRelationshipService(db.NewRelRepo(pool), onto), provisioning, seeds) } // testAuthToken is the static bearer token devConfig() configures. There is diff --git a/internal/httpapi/approvals.go b/internal/httpapi/approvals.go index 8bce52fa..78724daf 100644 --- a/internal/httpapi/approvals.go +++ b/internal/httpapi/approvals.go @@ -179,7 +179,7 @@ func (s *Server) DecideApproval(ctx context.Context, req gen.DecideApprovalReque _ = tx.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", targetID).Scan(&targetSlug) safego.Go("httpapi:executeApprovedAction", func() { - executeApprovedAction(context.Background(), s.pool, execID, targetSlug, actionStr) + s.executeApprovedAction(context.Background(), s.pool, execID, targetSlug, actionStr) }) // Status only — risk_class was set correctly at request time // (e.g. by policy.ClassifyCommand for `run`); overwriting it to diff --git a/internal/httpapi/fleet_health.go b/internal/httpapi/fleet_health.go index 54ea271d..ba6cdeee 100644 --- a/internal/httpapi/fleet_health.go +++ b/internal/httpapi/fleet_health.go @@ -4,7 +4,6 @@ import ( "context" "time" - "github.com/dtoro/oikos/internal/adapters/postgres" "github.com/dtoro/oikos/internal/httpapi/gen" ) @@ -68,7 +67,7 @@ func (s *Server) GetFleetHealth(ctx context.Context, req gen.GetFleetHealthReque } func (s *Server) ExportSeeds(ctx context.Context, req gen.ExportSeedsRequestObject) (gen.ExportSeedsResponseObject, error) { - exports, err := db.ExportToYAML(ctx, s.pool) + exports, err := s.seeds.Export(ctx) if err != nil { return nil, err } diff --git a/internal/httpapi/pct_create_test.go b/internal/httpapi/pct_create_test.go index 5198b613..78cc98b6 100644 --- a/internal/httpapi/pct_create_test.go +++ b/internal/httpapi/pct_create_test.go @@ -43,31 +43,6 @@ func TestFlexBoolUnmarshal(t *testing.T) { } } -func TestResolveTemplate(t *testing.T) { - avail := []string{ - "debian-12-standard_12.7-1_amd64.tar.zst", - "debian-13-standard_13.0-1_amd64.tar.zst", - "ubuntu-24.04-standard_24.04-2_amd64.tar.zst", - } - cases := []struct { - requested string - want string - }{ - {"debian-13-standard_13.0-1_amd64.tar.zst", "debian-13-standard_13.0-1_amd64.tar.zst"}, // exact - {"debian-13", "debian-13-standard_13.0-1_amd64.tar.zst"}, // prefix - {"", "debian-13-standard_13.0-1_amd64.tar.zst"}, // auto newest debian - {"debian-99", "debian-13-standard_13.0-1_amd64.tar.zst"}, // miss prefix → auto debian - } - for _, c := range cases { - if got := resolveTemplate(c.requested, avail); got != c.want { - t.Errorf("resolveTemplate(%q): got %q want %q", c.requested, got, c.want) - } - } - if got := resolveTemplate("debian-13", nil); got != "" { - t.Errorf("empty cache should yield empty, got %q", got) - } -} - // TestJSONErrValidForNastyOutput guards the bug where command output with // quotes/backslashes/newlines produced invalid JSON, failing the ::jsonb cast // and silently dropping the execution's final status update. @@ -87,32 +62,10 @@ func TestJSONErrValidForNastyOutput(t *testing.T) { } } -// TestGatewayPreflightPassed guards the exact bug found live: "UNREACHABLE" -// contains "REACHABLE" as a substring, so a strings.Contains(out,"REACHABLE") -// check is true for BOTH outcomes and can never fail. Exact-match only. -func TestGatewayPreflightPassed(t *testing.T) { - cases := []struct { - out string - want bool - }{ - {"PREFLIGHT_OK", true}, - {"PREFLIGHT_OK\n", true}, - {" PREFLIGHT_OK ", true}, - {"PREFLIGHT_FAIL", false}, - {"PREFLIGHT_FAIL\n", false}, - {"", false}, - {"some garbage output", false}, - // the specific historical bug: a naive substring check on the old - // REACHABLE/UNREACHABLE markers would have called this true. - {"UNREACHABLE", false}, - } - for _, c := range cases { - if got := gatewayPreflightPassed(c.out); got != c.want { - t.Errorf("gatewayPreflightPassed(%q) = %v, want %v", c.out, got, c.want) - } - } -} - +// resolveTemplate and gatewayPreflightPassed moved to the ssh provisioner +// adapter with the pct flow (Phase 7); their tests live in +// internal/adapters/ssh/provisioner_test.go. +// // provisionScript and sanitizePkgs were removed when pct_create was made // atomic (create + start + register only) — installing packages and running // setup scripts is now the agent's own job via follow-up `run` calls, which diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 48a694e5..538378d8 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -69,6 +69,11 @@ type Server struct { entityRepo *db.EntityRepo readModels ports.ReadModels relService *app.RelationshipService + // provisioning owns the pct_create flow (Phase 7): spec defaults, + // provisioner dispatch, guest registration in the graph. + provisioning *app.ProvisioningService + // seeds regenerates seed YAMLs for the export endpoint (Phase 7). + seeds *app.SeedService } // NewHandler builds the full HTTP handler: /healthz (unauthenticated, @@ -78,17 +83,19 @@ type Server struct { // holds a dedicated pooled connection for LISTEN. Callers MUST cancel ctx // before closing the pool — otherwise the held connection never releases // and pool.Close() deadlocks. -func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService) http.Handler { +func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService, provisioning *app.ProvisioningService, seeds *app.SeedService) http.Handler { s := &Server{ - pool: pool, - cfg: cfg, - entityCache: db.NewEntityCache(60 * time.Second), - sseBroker: newSSEBroker(10000), - sseSubs: make(map[*sseSubscriber]struct{}), - entities: entities, - entityRepo: entityRepo, - readModels: readModels, - relService: relService, + pool: pool, + cfg: cfg, + entityCache: db.NewEntityCache(60 * time.Second), + sseBroker: newSSEBroker(10000), + sseSubs: make(map[*sseSubscriber]struct{}), + entities: entities, + entityRepo: entityRepo, + readModels: readModels, + relService: relService, + provisioning: provisioning, + seeds: seeds, } // Wire secrets backend: Infisical primary with SOPS DR fallback. @@ -937,10 +944,10 @@ main(); // ListenAndServe runs the API server with graceful shutdown on ctx cancel // (SG4): stop accepting, drain in-flight for up to 30s, then exit. -func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService) error { +func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config, entities *app.EntityService, entityRepo *db.EntityRepo, readModels ports.ReadModels, relService *app.RelationshipService, provisioning *app.ProvisioningService, seeds *app.SeedService) error { srv := &http.Server{ Addr: cfg.APIListen, - Handler: NewHandler(ctx, pool, cfg, entities, entityRepo, readModels, relService), + Handler: NewHandler(ctx, pool, cfg, entities, entityRepo, readModels, relService, provisioning, seeds), ReadHeaderTimeout: 10 * time.Second, } diff --git a/internal/knowledge/seed_test.go b/internal/knowledge/seed_test.go deleted file mode 100644 index b0106fd5..00000000 --- a/internal/knowledge/seed_test.go +++ /dev/null @@ -1,155 +0,0 @@ -package knowledge - -import ( - "crypto/sha256" - "encoding/hex" - "reflect" - "testing" -) - -func TestContentHash(t *testing.T) { - t.Run("determinism", func(t *testing.T) { - a := contentHash("hello") - b := contentHash("hello") - if a != b { - t.Errorf("contentHash not deterministic: %q != %q", a, b) - } - }) - - t.Run("empty string known sha256", func(t *testing.T) { - got := contentHash("") - h := sha256.Sum256([]byte("")) - want := hex.EncodeToString(h[:]) - if got != want { - t.Errorf("contentHash(\"\") = %q, want %q", got, want) - } - }) - - t.Run("different inputs different outputs", func(t *testing.T) { - if contentHash("a") == contentHash("b") { - t.Error("different inputs produced same hash") - } - }) - - t.Run("output is 64-char hex", func(t *testing.T) { - got := contentHash("anything") - if len(got) != 64 { - t.Errorf("len = %d, want 64", len(got)) - } - for _, r := range got { - isHex := (r >= '0' && r <= '9') || (r >= 'a' && r <= 'f') - if !isHex { - t.Errorf("non-hex char %q in hash %q", r, got) - break - } - } - }) -} - -func TestStr(t *testing.T) { - cases := []struct { - name string - m map[string]any - key string - want string - }{ - {"missing key", map[string]any{}, "nope", ""}, - {"string value", map[string]any{"k": "v"}, "k", "v"}, - {"int value", map[string]any{"k": 42}, "k", ""}, - {"nil value", map[string]any{"k": nil}, "k", ""}, - {"empty string", map[string]any{"k": ""}, "k", ""}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - got := str(c.m, c.key) - if got != c.want { - t.Errorf("str() = %q, want %q", got, c.want) - } - }) - } -} - -func TestStrSlice(t *testing.T) { - cases := []struct { - name string - m map[string]any - key string - want []string - }{ - {"missing key", map[string]any{}, "tags", nil}, - {"all strings", map[string]any{"tags": []any{"a", "b", "c"}}, "tags", []string{"a", "b", "c"}}, - {"mixed types", map[string]any{"tags": []any{1, "a", true, "b"}}, "tags", []string{"a", "b"}}, - {"empty array", map[string]any{"tags": []any{}}, "tags", []string{}}, - {"nil elements filtered", map[string]any{"tags": []any{nil, "a", nil, "b"}}, "tags", []string{"a", "b"}}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - got := strSlice(c.m, c.key) - if len(got) != len(c.want) { - t.Errorf("len = %d, want %d (got %v)", len(got), len(c.want), got) - return - } - for i := range got { - if got[i] != c.want[i] { - t.Errorf("[%d] = %q, want %q", i, got[i], c.want[i]) - } - } - }) - } -} - -func TestMapVal(t *testing.T) { - cases := []struct { - name string - m map[string]any - key string - want map[string]any - }{ - {"missing key", map[string]any{}, "nope", nil}, - {"present map", map[string]any{"k": map[string]any{"x": 1}}, "k", map[string]any{"x": 1}}, - {"wrong type string", map[string]any{"k": "v"}, "k", nil}, - {"nested map", map[string]any{"k": map[string]any{"a": map[string]any{"b": 2}}}, "k", map[string]any{"a": map[string]any{"b": 2}}}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - got := mapVal(c.m, c.key) - if !reflect.DeepEqual(got, c.want) { - t.Errorf("mapVal() = %v, want %v", got, c.want) - } - }) - } -} - -func TestToPGArray(t *testing.T) { - cases := []struct { - name string - tags []string - want string - }{ - {"empty", []string{}, "{}"}, - {"single", []string{"a"}, `{"a"}`}, - {"multiple", []string{"a", "b"}, `{"a","b"}`}, - {"nil", nil, "{}"}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - got := toPGArray(c.tags) - if got != c.want { - t.Errorf("toPGArray() = %q, want %q", got, c.want) - } - }) - } - - // Special chars: tags containing " or \ are NOT escaped by toPGArray. - // This is a latent bug — Postgres array literals require these to be - // backslash-escaped. Test documents current behavior so a fix is - // detectable. Should be fixed. - t.Run("special chars unescaped (current buggy behavior)", func(t *testing.T) { - got := toPGArray([]string{`a"b`, `c\d`}) - // Current output: {"a"b","c\d"} — invalid Postgres array literal. - want := `{"a"b","c\d"}` - if got != want { - t.Errorf("toPGArray(special) = %q, want %q (if this changed, the escaping bug was fixed — update this test)", got, want) - } - }) -} diff --git a/plans/2026-08-15-hexagonal-architecture.md b/plans/2026-08-15-hexagonal-architecture.md index f74ae56c..09ed3f8a 100644 --- a/plans/2026-08-15-hexagonal-architecture.md +++ b/plans/2026-08-15-hexagonal-architecture.md @@ -1,7 +1,7 @@ # Hexagonal architecture for Oikos — design and phased refactor plan **Date:** 2026-08-15 -**Status:** Complete — Phases 0–9 shipped; all commits pushed to `main` +**Status:** In Progress — Phases 0–7 shipped (Phase 7 landed 2026-08-16: SeedService/SecretsService + ProvisioningService/Provisioner, CLI rewired). Phase 8 (nomos internal cleanup) not started. Phase 9 partially done (docs + version bump landed with the earlier commits; the ExecutionService/PolicyService ≥90% coverage gates and gating-matrix test, and the governance/execution slice they depend on, are still open — the shipped "Phase 4/5/6" commits converged entities, relationships, observation, signals, knowledge, and learning, not the §5 Phase-4 governance slice). **Scope:** All Go code (`cmd/oikos`, `cmd/nomos`, `cmd/webhook`) and the UI split. One hexagon covers the oikos backend; nomos is an external agent client that gets an internal cleanup (Phase 8) but stays outside the core. diff --git a/plans/index.md b/plans/index.md index 2039cb0f..e924e340 100644 --- a/plans/index.md +++ b/plans/index.md @@ -22,7 +22,7 @@ went sideways, open an investigation. | 2026-08-04 | [Hermes MCP client integration](done/2026-08-04-hermes-mcp-client-integration.md) | Done — deployed | | 2026-08-05 | [Agent execution safety: QEMU guest agent gate + host-mutation guard](done/2026-08-05-agent-execution-safety-qemu-guest-agent-gate.md) | Done — implemented (1b9c761) | | 2026-08-05 | [Backend evaluation: architecture, security, and reliability improvements](2026-08-05-backend-evaluation-improvements.md) | Done — all three phases (B, D, E) implemented as code (0.28.0–0.29.0), deployed, and hardened via review. Remaining: C (security) and F (performance) backlog. | -| 2026-08-15 | [Hexagonal architecture — design and phased refactor](2026-08-15-hexagonal-architecture.md) | Complete — all 10 phases shipped | +| 2026-08-15 | [Hexagonal architecture — design and phased refactor](2026-08-15-hexagonal-architecture.md) | In Progress — Phases 0–7 shipped; Phase 8 (nomos cleanup) and the Phase 9 coverage gates open | ## Done