feat: Phase 7 — SeedService, SecretsService, ProvisioningService + ssh Provisioner
Seed ingest/export moves behind ports.SeedRepository (SeedRepo in the postgres adapter; knowledge ingest absorbed from internal/knowledge, package deleted). pct_create flow (defaults, template/VMID/gateway pre-flights, pct create, graph registration) moves from httpapi's approved-execution path into app.ProvisioningService + the ssh provisioner adapter; CLI seed/export/secret become adapters over the services. EntityCreateInput gains EnrolledAt. Plan status corrected: phases 0-7 shipped, 8 + 9 gates open. VERSION 0.34.1.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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 <get|set|list|migrate|export-sops>")
|
||||
fmt.Fprintln(os.Stderr, "usage: oikos secret <get|set|list|verify|audit|migrate|export-sops>")
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
65
internal/adapters/postgres/seedrepo_test.go
Normal file
65
internal/adapters/postgres/seedrepo_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
237
internal/adapters/ssh/provisioner.go
Normal file
237
internal/adapters/ssh/provisioner.go
Normal file
@@ -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 <bridge>`) matters: a bare `ping <gw>` 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"
|
||||
}
|
||||
100
internal/adapters/ssh/provisioner_test.go
Normal file
100
internal/adapters/ssh/provisioner_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
239
internal/core/app/provisioning.go
Normal file
239
internal/core/app/provisioning.go
Normal file
@@ -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:<slug>) 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)
|
||||
}
|
||||
125
internal/core/app/provisioning_test.go
Normal file
125
internal/core/app/provisioning_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
42
internal/core/app/secrets.go
Normal file
42
internal/core/app/secrets.go
Normal file
@@ -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()
|
||||
}
|
||||
95
internal/core/app/seed.go
Normal file
95
internal/core/app/seed.go
Normal file
@@ -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)
|
||||
}
|
||||
98
internal/core/app/seed_test.go
Normal file
98
internal/core/app/seed_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
43
internal/core/ports/seeds.go
Normal file
43
internal/core/ports/seeds.go
Normal file
@@ -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)
|
||||
}
|
||||
@@ -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 <bridge>`) matters and was found
|
||||
// live: a plain unqualified `ping <gw>` 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:<hostname> 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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user