diff --git a/internal/checkdefaults/defaults.go b/internal/checkdefaults/defaults.go index 690e10f..39b2bb1 100644 --- a/internal/checkdefaults/defaults.go +++ b/internal/checkdefaults/defaults.go @@ -1,25 +1,369 @@ +// Package checkdefaults derives an entity's default check_defs from the +// monitoring kinds its type declares in seeds/ontology.yaml. +// +// The type says WHAT to watch (`service: [http, process]`); this package +// works out HOW — which concrete check_defs rows to write, and what host, +// script or URL each needs. Deriving config here rather than in YAML keeps +// the ontology declarative and keeps address resolution (which has to walk +// the graph) in code. package checkdefaults import ( "context" "encoding/json" "fmt" + "log/slog" + "strings" + "github.com/dtoro/oikos/internal/ontology" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) -type CheckDef struct { - Kind string - Script string - Host string - User string - Port int - Thresholds map[string]any - Extra map[string]any +// Semantic monitoring kinds, as declared on entity types. These are not +// check_defs.kind values — one semantic kind can expand to several concrete +// checks (`resource` becomes four ssh-script rows). +const ( + KindPing = "ping" + KindResource = "resource" + KindUpdates = "updates" + KindProcess = "process" + KindHTTP = "http" + KindCapacity = "capacity" + KindBackup = "backup-freshness" +) + +// defaultBackupMaxAge is how long a backup target may go without a new +// artifact before it is stale. A day suits the nightly jobs in this lab; +// override per target with `backup_max_age_s` in the entity's attributes. +const defaultBackupMaxAge = 86400 + +// Target is the entity default checks are being ensured for. +type Target struct { + ID uuid.UUID + Slug string + Type string + // Name is the entity's name column, not an attribute. The old code read + // attrs["name"], which is never populated — seeds put `name` beside + // `attributes`, not inside it — so every service silently produced no + // process check. + Name string + Attrs []byte +} + +// Result reports what Ensure did, so callers can log a type that declared +// monitoring but produced nothing instead of failing silently. +type Result struct { + Created int + // Skipped records kinds that were declared but could not be built, with + // the reason. A non-empty Skipped on an active entity is a real gap. + Skipped []Skip + // Undeclared is true when no ancestor of the type declared monitoring — + // an ontology gap rather than a fleet gap. + Undeclared bool +} + +// Skip is one declared-but-unbuilt check kind. +type Skip struct { + Kind string + Reason string +} + +type checkDef struct { + kind string + config map[string]any + interval int32 +} + +// Ensure writes the default check_defs for one entity, idempotently. +// +// Returns the number of checks created. An entity whose type declares +// monitoring it cannot satisfy comes back with a populated Skipped rather +// than an error — a missing address is a modelling gap, not a failure of +// this call. +func Ensure(ctx context.Context, tx pgx.Tx, tree *ontology.TypeTree, t Target) (Result, error) { + var res Result + + if _, err := tx.Exec(ctx, + `INSERT INTO entity_status (entity_id, health, updated_at) + VALUES ($1, 'unknown', now()) + ON CONFLICT (entity_id) DO NOTHING`, t.ID); err != nil { + return res, fmt.Errorf("entity_status %s: %w", t.Slug, err) + } + + mon := tree.Monitoring(t.Type) + if !mon.Declared { + res.Undeclared = true + return res, nil + } + if mon.None() { + return res, nil + } + + var attrs map[string]any + if len(t.Attrs) > 0 { + _ = json.Unmarshal(t.Attrs, &attrs) + } + if attrs == nil { + attrs = map[string]any{} + } + + // A service has no address of its own — it lives on the container that + // provides it. Fall back to the graph before giving up. + host := resolveHost(attrs) + if host == "" { + hostAttrs, err := hostViaGraph(ctx, tx, t.ID) + if err != nil { + return res, fmt.Errorf("resolve host for %s: %w", t.Slug, err) + } + host = resolveHost(hostAttrs) + if user := resolveSSHUser(hostAttrs); host != "" && user != "root" { + attrs["ssh"] = hostAttrs["ssh"] + } + } + user := resolveSSHUser(attrs) + port := resolveSSHPort(attrs) + + var defs []checkDef + for _, kind := range mon.Kinds { + built, reason := buildKind(kind, t, attrs, host, user, port) + if len(built) == 0 { + res.Skipped = append(res.Skipped, Skip{Kind: kind, Reason: reason}) + continue + } + defs = append(defs, built...) + } + + for i, def := range defs { + created, err := writeCheck(ctx, tx, t, i, def) + if err != nil { + return res, fmt.Errorf("check %s/%s: %w", t.Slug, def.kind, err) + } + if created { + res.Created++ + } + } + return res, nil +} + +// buildKind turns one declared semantic kind into concrete check_defs, or +// returns the reason it could not. +func buildKind(kind string, t Target, attrs map[string]any, host, user string, port int) ([]checkDef, string) { + ssh := func(script string, args ...string) checkDef { + cfg := map[string]any{"script": script, "host": host} + if user != "" && user != "root" { + cfg["user"] = user + } + if port != 0 && port != 22 { + cfg["port"] = port + } + if len(args) > 0 && args[0] != "" { + cfg["args"] = args[0] + } + return checkDef{kind: "ssh-script", config: cfg, interval: 60} + } + + switch kind { + case KindPing: + if host == "" { + return nil, "no address on the entity or its host" + } + return []checkDef{{kind: "ping", config: map[string]any{"host": host}, interval: 30}}, "" + + case KindResource: + if host == "" { + return nil, "no address on the entity or its host" + } + return []checkDef{ + ssh("cpu_check.sh"), ssh("memory_check.sh"), + ssh("load_check.sh"), ssh("disk_usage_check.sh"), + }, "" + + case KindUpdates: + if host == "" { + return nil, "no address on the entity or its host" + } + return []checkDef{ssh("updates_check.sh")}, "" + + case KindCapacity: + if host == "" { + return nil, "no address on the entity or its host" + } + return []checkDef{ssh("disk_usage_check.sh")}, "" + + case KindProcess: + if host == "" { + return nil, "no address on the entity or its host" + } + if t.Name == "" { + return nil, "no name to check a process for" + } + // process_check.sh takes the unit name as $1 and reports "unknown" + // without it. + return []checkDef{ssh("process_check.sh", t.Name)}, "" + + case KindBackup: + // A backup target is checked from the machine that writes to it, so it + // needs both an address (resolved via the backs-up-to edge) and the + // path to look at. + path, _ := attrs["path"].(string) + if path == "" { + return nil, "entity carries no path attribute to check for backups" + } + if host == "" { + return nil, "no address on the entity or whatever backs up to it" + } + maxAge := defaultBackupMaxAge + if v, ok := attrs["backup_max_age_s"].(float64); ok && v > 0 { + maxAge = int(v) + } + cfg := map[string]any{"path": path, "host": host, "max_age_s": maxAge} + if user != "" && user != "root" { + cfg["user"] = user + } + if port != 0 && port != 22 { + cfg["port"] = port + } + // Hourly: a daily backup does not need a 60s probe, and each one is an + // SSH round trip. + return []checkDef{{kind: "backup-freshness", config: cfg, interval: 3600}}, "" + + case KindHTTP: + url := httpURL(t, attrs) + if url == "" { + return nil, "no url attribute, public_host, or hostname-shaped name" + } + // max_status rather than an exact expected_status: most services sit + // behind Authentik and answer 302/401, which is a working service. + return []checkDef{{ + kind: "http", + config: map[string]any{"url": url, "max_status": 500}, + interval: 60, + }}, "" + } + + return nil, "no builder for this kind yet" +} + +// writeCheck upserts one check_def and its backing check entity. +// +// The entity upsert MUST return the row's id. The previous version generated +// a fresh uuid, inserted ON CONFLICT (slug) DO NOTHING, then wrote a +// check_defs row referencing that uuid. On any re-seed the slug already +// existed, the entity insert became a no-op, and the check_defs insert +// violated its foreign key — which aborted the whole ingest transaction and +// made every subsequent statement fail with 25P02. Because the errors were +// discarded, the only visible symptom was an unrelated failure much later. +func writeCheck(ctx context.Context, tx pgx.Tx, t Target, idx int, def checkDef) (bool, error) { + // The full target slug, not a truncation of it. shortSlug() took the last + // 8 characters, so all 21 ingress routes collapsed to ".network" and + // generated one identical check slug — they overwrote each other and 20 + // of them ended up with no check at all. It also collided service:jellyfin + // with lxc:jellyfin. Entity slugs are unique; use them. + checkSlug := fmt.Sprintf("check:%s:%s:%d", def.kind, t.Slug, idx) + + newID, err := uuid.NewV7() + if err != nil { + newID = uuid.New() + } + + var checkID uuid.UUID + err = tx.QueryRow(ctx, + `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) + VALUES ($1, $2, 'check', $2, 'active', '{}', 1, now(), now()) + ON CONFLICT (slug) DO UPDATE SET updated_at = now() + RETURNING id`, + newID, checkSlug).Scan(&checkID) + if err != nil { + return false, fmt.Errorf("upsert check entity %s: %w", checkSlug, err) + } + + configJSON, err := json.Marshal(def.config) + if err != nil { + return false, err + } + + // Config is derived from the seed, so the seed wins on re-ingest and + // attribute changes propagate. `enabled` is deliberately left alone: it + // is operational state an operator may have toggled. + tag, err := tx.Exec(ctx, + `INSERT INTO check_defs (entity_id, target_id, kind, config, interval_s, timeout_s, enabled) + VALUES ($1, $2, $3, $4, $5, 30, true) + ON CONFLICT (entity_id) DO UPDATE + SET target_id = EXCLUDED.target_id, kind = EXCLUDED.kind, + config = EXCLUDED.config, interval_s = EXCLUDED.interval_s, + updated_at = now()`, + checkID, t.ID, def.kind, configJSON, def.interval) + if err != nil { + return false, fmt.Errorf("upsert check_def %s: %w", checkSlug, err) + } + return tag.RowsAffected() > 0, nil +} + +// httpURL works out what to GET for an http check. +// +// Ingress routes carry their hostname as the entity name rather than as an +// attribute (`name: media.hubris.network`), and most declare no attributes at +// all — so the name is the only thing to go on. Requiring a `url` attribute +// left all 21 of them unmonitored, which is a shame given an ingress check is +// the most end-to-end probe available: it exercises Caddy, DNS, TLS and the +// upstream in one request. +func httpURL(t Target, attrs map[string]any) string { + if url, ok := attrs["url"].(string); ok && url != "" { + return url + } + if h, ok := attrs["public_host"].(string); ok && h != "" { + return "https://" + h + } + // A dotted name is a hostname; a service name like "jellyfin" is not. + if strings.Contains(t.Name, ".") && !strings.Contains(t.Name, " ") { + return "https://" + t.Name + } + return "" +} + +// hostViaGraph returns the attributes of the entity that hosts or provides +// this one, so a service can inherit its container's address. +func hostViaGraph(ctx context.Context, tx pgx.Tx, entityID uuid.UUID) (map[string]any, error) { + rows, err := tx.Query(ctx, ` + SELECT e.attributes + FROM relationships r + JOIN entities e ON e.id = r.source_id + WHERE r.target_id = $1 + AND r.valid_to IS NULL + -- backs-up-to points from the thing being backed up TO the target, + -- so walking it backwards finds the machine that writes the backups + -- — which is the only place a freshness check can run. + AND r.type IN ('provides', 'hosts', 'runs-on', 'backs-up-to') + ORDER BY CASE r.type + WHEN 'provides' THEN 0 WHEN 'runs-on' THEN 1 + WHEN 'backs-up-to' THEN 2 ELSE 3 END`, + entityID) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var raw []byte + if err := rows.Scan(&raw); err != nil { + return nil, err + } + var attrs map[string]any + if json.Unmarshal(raw, &attrs) != nil { + continue + } + if resolveHost(attrs) != "" { + return attrs, nil + } + } + return nil, rows.Err() } func resolveHost(attrs map[string]any) string { + if attrs == nil { + return "" + } if ip, ok := attrs["lan_ip"].(string); ok && ip != "" { return ip } @@ -28,11 +372,21 @@ func resolveHost(attrs map[string]any) string { if ip, ok := nb["ip"].(string); ok && ip != "" { return ip } + // Seeds record the mesh name, not an address — ws:mac-mini + // carries only `fqdn`, which is why it resolved to nothing. + if fqdn, ok := nb["fqdn"].(string); ok && fqdn != "" { + return fqdn + } } } if ip, ok := attrs["mesh_ip"].(string); ok && ip != "" { return ip } + for _, key := range []string{"host", "address", "public_host"} { + if v, ok := attrs[key].(string); ok && v != "" { + return v + } + } return "" } @@ -57,148 +411,17 @@ func resolveSSHPort(attrs map[string]any) int { return 22 } -func forEntityType(entityType string, attrs map[string]any) []CheckDef { - host := resolveHost(attrs) - user := resolveSSHUser(attrs) - port := resolveSSHPort(attrs) - - ssh := func(script string) CheckDef { - return CheckDef{Kind: "ssh-script", Script: script, Host: host, User: user, Port: port} - } - - switch entityType { - case "proxmox-host", "standalone-server": - if host == "" { - return nil +// LogResult emits the one line that was missing: a type that asked for +// monitoring and did not get it. +func LogResult(slug, entityType string, res Result) { + switch { + case res.Undeclared: + slog.Info("checkdefaults: type declares no monitoring", + "entity", slug, "type", entityType) + case len(res.Skipped) > 0: + for _, s := range res.Skipped { + slog.Warn("checkdefaults: declared check not created", + "entity", slug, "type", entityType, "kind", s.Kind, "reason", s.Reason) } - return []CheckDef{ - {Kind: "ping", Host: host}, - ssh("cpu_check.sh"), - ssh("memory_check.sh"), - ssh("load_check.sh"), - ssh("disk_usage_check.sh"), - ssh("updates_check.sh"), - } - case "workstation": - if host == "" { - return nil - } - return []CheckDef{ - {Kind: "ping", Host: host}, - ssh("cpu_check.sh"), - ssh("memory_check.sh"), - ssh("load_check.sh"), - } - case "lxc": - if host == "" { - return nil - } - return []CheckDef{ - ssh("cpu_check.sh"), - ssh("memory_check.sh"), - ssh("load_check.sh"), - ssh("disk_usage_check.sh"), - } - case "vm": - if host == "" { - return nil - } - return []CheckDef{ - {Kind: "ping", Host: host}, - } - case "service": - if host == "" { - return nil - } - n, _ := attrs["name"].(string) - if n == "" { - return nil - } - return []CheckDef{ - {Kind: "ssh-script", Script: "process_check.sh", Host: host, User: user, Port: port, - Extra: map[string]any{"args": n}}, - } - } - return nil -} - -func shortSlug(slug string) string { - const n = 8 - if len(slug) > n { - return slug[len(slug)-n:] - } - return slug -} - -func defaultInterval(kind string) int32 { - switch kind { - case "ping": - return 30 - case "ssh-script": - return 60 - default: - return 300 - } -} - -func Ensure(ctx context.Context, tx pgx.Tx, entityID uuid.UUID, slug, entityType string, attrsJSON []byte) { - _, _ = tx.Exec(ctx, - `INSERT INTO entity_status (entity_id, health, updated_at) - VALUES ($1, 'unknown', now()) - ON CONFLICT (entity_id) DO NOTHING`, - entityID) - - var attrs map[string]any - if len(attrsJSON) > 0 { - json.Unmarshal(attrsJSON, &attrs) - } - if attrs == nil { - attrs = map[string]any{} - } - - defs := forEntityType(entityType, attrs) - if len(defs) == 0 { - return - } - - for i, def := range defs { - checkID, err := uuid.NewV7() - if err != nil { - checkID = uuid.New() - } - checkSlug := fmt.Sprintf("check:%s:%s:%d", def.Kind, shortSlug(slug), i) - - _, _ = tx.Exec(ctx, - `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) - VALUES ($1, $2, 'check', $2, 'active', '{}', 1, now(), now()) - ON CONFLICT (slug) DO NOTHING`, - checkID, checkSlug) - - configMap := map[string]any{} - if def.Script != "" { - configMap["script"] = def.Script - } - if def.Host != "" { - configMap["host"] = def.Host - } - if def.User != "" && def.User != "root" { - configMap["user"] = def.User - } - if def.Port != 0 && def.Port != 22 { - configMap["port"] = def.Port - } - if def.Thresholds != nil { - configMap["thresholds"] = def.Thresholds - } - for k, v := range def.Extra { - configMap[k] = v - } - configJSON, _ := json.Marshal(configMap) - - _, _ = tx.Exec(ctx, - `INSERT INTO check_defs (entity_id, target_id, kind, config, interval_s, timeout_s, enabled) - VALUES ($1, $2, $3, $4, $5, 30, true) - ON CONFLICT (entity_id) DO NOTHING`, - checkID, entityID, def.Kind, configJSON, defaultInterval(def.Kind)) } } diff --git a/internal/checkdefaults/defaults_test.go b/internal/checkdefaults/defaults_test.go new file mode 100644 index 0000000..c826359 --- /dev/null +++ b/internal/checkdefaults/defaults_test.go @@ -0,0 +1,125 @@ +package checkdefaults + +import ( + "testing" +) + +// The attribute shapes here are copied from seeds/inventory.yaml. The original +// resolveHost looked for lan_ip / mesh.netbird.ip / mesh_ip, none of which a +// service or workstation actually carries — which is why 86 of 89 entities +// ended up with no checks. +func TestResolveHostAcceptsRealSeedShapes(t *testing.T) { + cases := []struct { + desc string + attrs map[string]any + want string + }{ + {"lxc carries lan_ip", map[string]any{"lan_ip": "192.168.8.246"}, "192.168.8.246"}, + { + "ws:mac-mini carries only a netbird fqdn", + map[string]any{"mesh": map[string]any{"netbird": map[string]any{ + "fqdn": "mac-mini-234-17.netbird.selfhosted"}}}, + "mac-mini-234-17.netbird.selfhosted", + }, + { + "a netbird ip still wins over the fqdn", + map[string]any{"mesh": map[string]any{"netbird": map[string]any{ + "ip": "100.122.0.10", "fqdn": "x.netbird.selfhosted"}}}, + "100.122.0.10", + }, + {"public_host as a last resort", map[string]any{"public_host": "media.hubris.network"}, "media.hubris.network"}, + {"a service carries no address at all", map[string]any{ + "url": "https://media.hubris.network", "port": 8096}, ""}, + {"nil attrs", nil, ""}, + } + + for _, c := range cases { + if got := resolveHost(c.attrs); got != c.want { + t.Errorf("%s: resolveHost = %q, want %q", c.desc, got, c.want) + } + } +} + +func TestHTTPURLPrefersAttributeThenName(t *testing.T) { + cases := []struct { + desc string + name string + attrs map[string]any + want string + }{ + {"explicit url wins", "jellyfin", + map[string]any{"url": "https://media.hubris.network"}, "https://media.hubris.network"}, + {"public_host becomes https", "jellyfin", + map[string]any{"public_host": "media.hubris.network"}, "https://media.hubris.network"}, + // Ingress routes carry the hostname as the entity name and usually + // declare no attributes at all. + {"hostname-shaped name", "media.hubris.network", nil, "https://media.hubris.network"}, + {"a bare service name is not a hostname", "jellyfin", nil, ""}, + } + + for _, c := range cases { + got := httpURL(Target{Name: c.name}, c.attrs) + if got != c.want { + t.Errorf("%s: httpURL = %q, want %q", c.desc, got, c.want) + } + } +} + +func TestBuildKindReportsWhyItSkipped(t *testing.T) { + // A declared kind that cannot be built must explain itself rather than + // vanish — that silence is what hid the coverage gap. + if defs, reason := buildKind(KindPing, Target{}, nil, "", "root", 22); len(defs) != 0 || reason == "" { + t.Errorf("ping without a host should skip with a reason, got %d defs / %q", len(defs), reason) + } + if defs, reason := buildKind(KindProcess, Target{Name: ""}, nil, "10.0.0.1", "root", 22); len(defs) != 0 || reason == "" { + t.Errorf("process without a name should skip with a reason, got %d defs / %q", len(defs), reason) + } + if defs, reason := buildKind("dns", Target{}, nil, "10.0.0.1", "root", 22); len(defs) != 0 || reason == "" { + t.Errorf("an unimplemented kind should skip with a reason, got %d defs / %q", len(defs), reason) + } +} + +func TestBuildKindProcessPassesTheUnitName(t *testing.T) { + // process_check.sh reads $1 and answers "no service name provided" + // without it. checkdefaults always wrote args; nothing read them. + defs, reason := buildKind(KindProcess, Target{Name: "jellyfin"}, nil, "10.0.0.1", "root", 22) + if len(defs) != 1 { + t.Fatalf("expected one process check, got %d (%s)", len(defs), reason) + } + if got := defs[0].config["args"]; got != "jellyfin" { + t.Errorf("process check args = %v, want jellyfin", got) + } + if got := defs[0].config["script"]; got != "process_check.sh" { + t.Errorf("process check script = %v", got) + } +} + +func TestBuildKindHTTPUsesAStatusRangeNotAnExactCode(t *testing.T) { + // Most services sit behind Authentik and answer 302/401. + defs, _ := buildKind(KindHTTP, Target{Name: "jellyfin"}, + map[string]any{"url": "https://media.hubris.network"}, "", "root", 22) + if len(defs) != 1 { + t.Fatalf("expected one http check, got %d", len(defs)) + } + if got := defs[0].config["max_status"]; got != 500 { + t.Errorf("max_status = %v, want 500", got) + } + if _, exact := defs[0].config["expected_status"]; exact { + t.Error("default http checks must not pin an exact status") + } +} + +func TestBuildKindResourceExpandsToFourScripts(t *testing.T) { + defs, _ := buildKind(KindResource, Target{}, nil, "10.0.0.1", "root", 22) + if len(defs) != 4 { + t.Fatalf("resource should expand to 4 checks, got %d", len(defs)) + } + for _, d := range defs { + if d.kind != "ssh-script" { + t.Errorf("resource check kind = %q, want ssh-script", d.kind) + } + if d.config["host"] != "10.0.0.1" { + t.Errorf("resource check lost its host: %v", d.config) + } + } +} diff --git a/internal/db/integration_test.go b/internal/db/integration_test.go index 7a103bf..5a4ca7b 100644 --- a/internal/db/integration_test.go +++ b/internal/db/integration_test.go @@ -167,6 +167,62 @@ func TestSeedIngestIdempotentAndNoDuplicateEdges(t *testing.T) { } } +// Regression: insertOneEntityType read tMap["attribute_schema"], but +// seeds/ontology.yaml spells the key `attributes:`. The mismatch marshalled a +// nil into the JSON literal `null` for every one of the 60 types, so no +// attribute schema was ever ingested — the API and `oikos export` returned +// null across the board, silently, for the life of the project. +func TestSeedIngestsAttributeSchemas(t *testing.T) { + pool := newTestPool(t) + seedAll(t, pool, seedsDir()) + + if n := count(t, pool, + `SELECT count(*) FROM entity_types WHERE attribute_schema = 'null'::jsonb`); n != 0 { + t.Errorf("%d entity types stored the JSON literal null instead of a schema or SQL NULL", n) + } + + if n := count(t, pool, + `SELECT count(*) FROM entity_types WHERE jsonb_typeof(attribute_schema) = 'object'`); n == 0 { + t.Fatal("no entity type ingested an attribute schema") + } + + // A type declaring `attributes:` must round-trip its properties. + if n := count(t, pool, `SELECT count(*) FROM entity_types + WHERE name = 'lxc' AND attribute_schema #>> '{properties,pve_id,type}' = 'integer'`); n != 1 { + t.Error("lxc.attribute_schema lost its declared pve_id property") + } + + // A type declaring none stores SQL NULL, not a JSON null. + if n := count(t, pool, `SELECT count(*) FROM entity_types + WHERE name = 'sensor' AND attribute_schema IS NULL`); n != 1 { + t.Error("a type declaring no attributes should store SQL NULL") + } +} + +// monitoring_spec drives which entities coverageSweep may flag as unmonitored, +// so the three states have to survive ingest distinctly: SQL NULL (undeclared, +// resolved from an ancestor or the layer default), '[]' (explicitly +// unmonitorable), and a non-empty array (the kinds the type warrants). +func TestSeedIngestsMonitoringSpec(t *testing.T) { + pool := newTestPool(t) + seedAll(t, pool, seedsDir()) + + cases := []struct { + typ, where, desc string + }{ + {"service", `monitoring_spec = '["http","process"]'::jsonb`, "declared kinds"}, + {"machine", `monitoring_spec = '["ping","resource","updates"]'::jsonb`, "declared on an abstract type"}, + {"site", `monitoring_spec = '[]'::jsonb`, "explicitly unmonitorable"}, + {"lxc", `monitoring_spec IS NULL`, "inherits from container, so its own column is NULL"}, + } + for _, c := range cases { + if n := count(t, pool, fmt.Sprintf( + `SELECT count(*) FROM entity_types WHERE name = '%s' AND %s`, c.typ, c.where)); n != 1 { + t.Errorf("%s (%s): monitoring_spec did not match %s", c.typ, c.desc, c.where) + } + } +} + func TestAbstractTypeRejected(t *testing.T) { pool := newTestPool(t) seedAll(t, pool, seedsDir()) diff --git a/internal/db/seed.go b/internal/db/seed.go index e93a1d6..1df7c2a 100644 --- a/internal/db/seed.go +++ b/internal/db/seed.go @@ -21,6 +21,7 @@ type SeedResult struct { RiskClasses int ApprovalRules int AutonomySettings int + Checks int } // IngestOntologySeed ingests seeds/ontology.yaml into the DB. @@ -96,6 +97,7 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (* // Entities entities, _ := data["entities"].([]any) entityTypes := make(map[string]string) // slug -> type, for edge validation + var pendingChecks []checkdefaults.Target for _, raw := range entities { eMap, ok := raw.(map[string]any) if !ok { @@ -144,7 +146,12 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (* return nil, fmt.Errorf("entity_status %s: %w", slug, err) } - checkdefaults.Ensure(ctx, tx, entityID, slug, typeName, attrsBytes) + // Default checks are deferred until after relationships are ingested: + // a service has no address of its own and inherits its container's, + // which means the hosting edge has to exist first. + pendingChecks = append(pendingChecks, checkdefaults.Target{ + ID: entityID, Slug: slug, Type: typeName, Name: name, Attrs: attrsBytes, + }) r.Entities++ } @@ -208,6 +215,19 @@ func IngestInventorySeed(ctx context.Context, tx pgx.Tx, data map[string]any) (* return nil, err } + // Default checks, now that hosting edges exist. Errors here are fatal: + // swallowing them is what let a foreign-key violation abort the ingest + // transaction while surfacing as an unrelated failure several entities + // later. + for _, target := range pendingChecks { + res, err := checkdefaults.Ensure(ctx, tx, tree, target) + if err != nil { + return nil, fmt.Errorf("default checks for %s: %w", target.Slug, err) + } + checkdefaults.LogResult(target.Slug, target.Type, res) + r.Checks += res.Created + } + return r, nil } @@ -361,19 +381,68 @@ func insertOneEntityType(ctx context.Context, tx pgx.Tx, name string, tMap map[s layer, _ := tMap["layer"].(string) desc, _ := tMap["description"].(string) lifecycleID, _ := tMap["lifecycle"].(string) - attrSchema := tMap["attribute_schema"] - schemaBytes, _ := json.Marshal(attrSchema) + // seeds/ontology.yaml spells this `attributes:`. Reading it as + // "attribute_schema" silently marshalled nil to the JSON literal `null` + // for every type, so no attribute schema was ever ingested — the API and + // `oikos export` returned null for all 60 types. + attrSchema := tMap["attributes"] _, err := tx.Exec(ctx, `INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, description, - lifecycle_id, attribute_schema, schema_version, status, created_at, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 1, 'active', now(), now()) + lifecycle_id, attribute_schema, monitoring_spec, schema_version, status, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 1, 'active', now(), now()) ON CONFLICT (name) DO UPDATE SET parent_type = $2, is_abstract = $3, domain = $4, - layer = $5, description = $6, lifecycle_id = $7, attribute_schema = $8, updated_at = now()`, - name, nullableStr(parent), isAbstract, domain, layer, desc, nullableStr(lifecycleID), nullableStr(string(schemaBytes))) + layer = $5, description = $6, lifecycle_id = $7, attribute_schema = $8, + monitoring_spec = $9, updated_at = now()`, + name, nullableStr(parent), isAbstract, domain, layer, desc, nullableStr(lifecycleID), + attributeSchemaJSON(attrSchema), monitoringSpecJSON(tMap["monitoring"])) return err } +// attributeSchemaJSON marshals a type's `attributes:` block for storage, +// mapping "the type declares no schema" to SQL NULL rather than to the JSON +// literal `null`. Both readers already treat a JSON `null` as absent, but a +// real NULL is what `attribute_schema IS NULL` expects and is what the column +// meant all along. +func attributeSchemaJSON(v any) any { + if v == nil { + return nil + } + b, err := json.Marshal(v) + if err != nil { + return nil + } + return string(b) +} + +// monitoringSpecJSON normalises an entity type's `monitoring:` declaration into +// the JSONB stored in entity_types.monitoring_spec. Three outcomes, and the +// difference between the last two is load-bearing for coverage signalling: +// +// absent → nil (SQL NULL) — undeclared, an ontology gap +// none | [] → "[]" — explicitly unmonitorable, by design +// [http, resource]→ '["http","resource"]' +// +// `monitoring: none` is accepted as a more legible spelling of `[]`; YAML +// parses the bare word as the string "none", not as null. +func monitoringSpecJSON(v any) any { + switch spec := v.(type) { + case nil: + return nil + case string: + if spec == "none" { + return "[]" + } + // A single kind written unquoted, e.g. `monitoring: http`. + b, _ := json.Marshal([]string{spec}) + return string(b) + case []any: + b, _ := json.Marshal(toStringSlice(spec)) + return string(b) + } + return nil +} + func toStringSlice(v any) []string { if v == nil { return nil diff --git a/internal/db/typetree.go b/internal/db/typetree.go index c848ef1..36a4886 100644 --- a/internal/db/typetree.go +++ b/internal/db/typetree.go @@ -19,7 +19,8 @@ func LoadTypeTree(ctx context.Context, tx pgx.Tx) (*ontology.TypeTree, error) { } rows, err := tx.Query(ctx, - `SELECT name, COALESCE(parent_type,''), is_abstract, COALESCE(lifecycle_id,'') + `SELECT name, COALESCE(parent_type,''), is_abstract, COALESCE(lifecycle_id,''), + layer, monitoring_spec FROM entity_types`) if err != nil { return nil, fmt.Errorf("load entity_types: %w", err) @@ -27,10 +28,16 @@ func LoadTypeTree(ctx context.Context, tx pgx.Tx) (*ontology.TypeTree, error) { for rows.Next() { var name string var info ontology.TypeInfo - if err := rows.Scan(&name, &info.Parent, &info.IsAbstract, &info.LifecycleID); err != nil { + // NULL monitoring_spec means the type declared nothing; '[]' means it + // declared "explicitly unmonitorable". Scanning through a pointer is + // what keeps those two apart — see ontology.TypeTree.Monitoring. + var monitoring *[]string + if err := rows.Scan(&name, &info.Parent, &info.IsAbstract, &info.LifecycleID, + &info.Layer, &monitoring); err != nil { rows.Close() return nil, err } + info.Monitoring = monitoring t.Types[name] = info } rows.Close() diff --git a/internal/execlog/execlog.go b/internal/execlog/execlog.go new file mode 100644 index 0000000..9529ee5 --- /dev/null +++ b/internal/execlog/execlog.go @@ -0,0 +1,135 @@ +// Package execlog persists incremental command output for an execution and +// announces it on the event stream. +// +// It exists as its own package because both SSH execution paths need it — +// internal/mcp (the agent's auto-run windows) and internal/httpapi (the +// post-approval actuator). Those two already carry near-identical copies of +// sshExec, and every bug found in this area so far has been a case of the two +// copies drifting apart; one shared sink is the cheap way not to repeat that. +package execlog + +import ( + "context" + "log/slog" + "sync" + "time" + + "github.com/dtoro/oikos/internal/db" + "github.com/dtoro/oikos/internal/db/sqlcgen" + "github.com/dtoro/oikos/internal/observability" + "github.com/google/uuid" +) + +// eventInterval throttles execution.output events. Chunks are persisted as +// they arrive, but a chatty command (apt, a long build) can produce hundreds +// per second and the SSE broker drops events for slow subscribers — flooding +// it would push out the signal.* and approval.* events that actually need to +// arrive. The event is only a "there is more output" ping; subscribers re-read +// the rows. +const eventInterval = time.Second + +// Sink receives output chunks as they arrive from a remote command. +type Sink func(stream string, chunk []byte) + +// New returns a Sink that writes chunks to execution_logs and emits a +// throttled execution.output event, plus a Flush to call when the command +// finishes. +// +// The returned Sink is safe for concurrent use: stdout and stderr are written +// from separate goroutines. +func New(ctx context.Context, pool *db.Pool, execID uuid.UUID, correlationID string) (Sink, func()) { + var ( + mu sync.Mutex + seq int + lastEvent time.Time + pending bool + ) + + emit := func() { + if err := observability.Event(ctx, sqlcgen.New(pool), "execution.output", &execID, + "info", "actuator", correlationID, map[string]any{"execution_id": execID.String()}); err != nil { + slog.Debug("execlog: emit output event", "error", err, "execution_id", execID) + } + } + + sink := func(stream string, chunk []byte) { + if len(chunk) == 0 { + return + } + mu.Lock() + seq++ + n := seq + mu.Unlock() + + // A failed log write must never fail the command: this is observability, + // and the authoritative output still lands in executions.result at the + // end. Log and carry on. + if _, err := pool.Exec(ctx, + `INSERT INTO execution_logs (execution_id, seq, stream, chunk) + VALUES ($1, $2, $3, $4)`, + execID, n, stream, string(chunk)); err != nil { + slog.Debug("execlog: persist chunk", "error", err, "execution_id", execID) + return + } + + mu.Lock() + due := time.Since(lastEvent) >= eventInterval + if due { + lastEvent = time.Now() + pending = false + } else { + pending = true + } + mu.Unlock() + + if due { + emit() + } + } + + // Flush emits a final event when output arrived inside the throttle window, + // so the last few lines of a short command are not left unannounced. + flush := func() { + mu.Lock() + due := pending + pending = false + mu.Unlock() + if due { + emit() + } + } + + return sink, flush +} + +// Read returns an execution's persisted output in order. +func Read(ctx context.Context, pool *db.Pool, execID uuid.UUID, limit int) ([]Chunk, error) { + if limit <= 0 { + limit = 1000 + } + rows, err := pool.Query(ctx, + `SELECT seq, stream, chunk, ts FROM execution_logs + WHERE execution_id = $1 ORDER BY seq LIMIT $2`, execID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []Chunk + for rows.Next() { + var c Chunk + if err := rows.Scan(&c.Seq, &c.Stream, &c.Chunk, &c.TS); err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + +// Chunk is one persisted slice of command output. +type Chunk struct { + Seq int `json:"seq"` + Stream string `json:"stream"` + Chunk string `json:"chunk"` + TS time.Time `json:"ts"` +} diff --git a/internal/httpapi/actuator.go b/internal/httpapi/actuator.go index 84dce48..e14574d 100644 --- a/internal/httpapi/actuator.go +++ b/internal/httpapi/actuator.go @@ -1,6 +1,7 @@ package httpapi import ( + "bytes" "context" "encoding/base64" "encoding/json" @@ -9,10 +10,12 @@ import ( "os" "strconv" "strings" + "sync" "time" "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/db/sqlcgen" + "github.com/dtoro/oikos/internal/execlog" "github.com/dtoro/oikos/internal/observability" "github.com/google/uuid" "golang.org/x/crypto/ssh" @@ -71,7 +74,35 @@ func initSSH() { // report. Generous enough for a real apt/docker install; not infinite. const sshExecTimeout = 10 * time.Minute +// streamWriter buffers everything it is given while forwarding each write to a +// sink. One on session.Stdout and another sharing the same buffer on +// session.Stderr reproduces CombinedOutput's interleaving in the order the +// remote end produced it. Mirrors the twin in internal/mcp/server.go. +type streamWriter struct { + mu *sync.Mutex + buf *bytes.Buffer + stream string + sink execlog.Sink +} + +func (w *streamWriter) Write(p []byte) (int, error) { + w.mu.Lock() + w.buf.Write(p) + w.mu.Unlock() + if w.sink != nil { + // Copy: the ssh library reuses p once Write returns. + w.sink(w.stream, append([]byte(nil), p...)) + } + return len(p), nil +} + 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) { initSSH() if len(_sshKey) == 0 { return "", fmt.Errorf("no SSH key available") @@ -105,50 +136,62 @@ func sshExec(ctx context.Context, host, user, command string) (string, error) { } defer session.Close() - type result struct { - out []byte - err error + var ( + mu sync.Mutex + buf bytes.Buffer + ) + session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink} + session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink} + + collected := func() string { + mu.Lock() + defer mu.Unlock() + return strings.TrimSpace(buf.String()) } - done := make(chan result, 1) + + done := make(chan error, 1) go func() { // See internal/mcp/server.go's sshExec for why this recovers rather // than letting a rare SSH-library panic crash the whole api process. defer func() { if r := recover(); r != nil { - done <- result{nil, fmt.Errorf("panic in ssh exec: %v", r)} + done <- fmt.Errorf("panic in ssh exec: %v", r) } }() - out, err := session.CombinedOutput(command) - done <- result{out, err} + // Run rather than CombinedOutput so the assigned writers are used; + // Run returns only after both streams are fully drained. + done <- session.Run(command) }() select { - case r := <-done: - text := strings.TrimSpace(string(r.out)) + case err := <-done: + text := collected() // A non-zero exit MUST surface as an error. The previous guard only // errored when there was no output, so a `pct create` that printed // "CT 132 already exists" and exited non-zero was reported as // success — the execution was marked completed though nothing was // provisioned. - if r.err != nil { + if err != nil { if text != "" { - return text, fmt.Errorf("%w: %s", r.err, text) + return text, fmt.Errorf("%w: %s", err, text) } - return text, fmt.Errorf("exec: %w", r.err) + return text, fmt.Errorf("exec: %w", err) } return text, nil case <-time.After(sshExecTimeout): // Close the session/client to hang up the remote side; the - // goroutine above will eventually exit once that unblocks - // CombinedOutput, but we don't wait for it — the caller needs an - // answer now, not an indefinite hang. + // goroutine above will eventually exit once that unblocks Run, but we + // don't wait for it — the caller needs an answer now, not an + // indefinite hang. session.Close() client.Close() - return "", fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host) + // Return what arrived before it hung, rather than "". A provisioning + // command that stalls halfway is precisely when its output matters. + return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host) case <-ctx.Done(): session.Close() client.Close() - return "", ctx.Err() + return collected(), ctx.Err() } } @@ -231,7 +274,16 @@ func emitExecutionEvent(ctx context.Context, pool *db.Pool, execID uuid.UUID, st if status == "failed" { severity = "warning" } - _ = observability.Event(ctx, sqlcgen.New(pool), "execution."+status, &execID, severity, "actuator", "", detail) + // The correlation id was hardcoded to "", so execution events could not be + // tied back to the session that caused them — the one join you want when + // asking "what did this agent turn actually do?". It is already on the + // execution row; read it rather than threading it through eleven callers. + var correlationID string + if err := pool.QueryRow(ctx, + `SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); err != nil { + correlationID = "" + } + _ = observability.Event(ctx, sqlcgen.New(pool), "execution."+status, &execID, severity, "actuator", correlationID, detail) if status == "completed" || status == "failed" || status == "cancelled" { closePlanStepForExecution(ctx, pool, execID, status) } @@ -280,6 +332,29 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, action, params := actionStr[:idx], actionStr[idx+1:] startedAt := time.Now() + // Persist started_at now, not at the end. It was captured here but only + // written in the terminal UPDATE, so a running execution reported + // started_at = NULL for its entire life — the UI could not show how long + // anything had been going, which is exactly when you want to know. + if _, err := pool.Exec(ctx, + `UPDATE executions SET status = 'running', started_at = $2 WHERE entity_id = $1`, + execID, startedAt); err != nil { + slog.Error("httpapi: mark execution running", "error", err, "execution_id", execID) + } + + // Stream output for the actions whose output an operator actually watches: + // a long apt upgrade, a pct create, an arbitrary approved `run`. The small + // internal lookups further down (listing template cache, pvesh nextid) stay + // unstreamed — they are plumbing, and logging them would bury the command + // the operator approved. + var correlationID string + if qerr := pool.QueryRow(ctx, + `SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); qerr != nil { + correlationID = "" + } + sink, flushLogs := execlog.New(ctx, pool, execID, correlationID) + defer flushLogs() + var output, cmd string switch action { @@ -295,30 +370,30 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, default: cmd = fmt.Sprintf("systemctl %s %s 2>&1", params, svc) } - output, err = sshExec(ctx, host, user, cmd) + output, err = sshExecStream(ctx, host, user, cmd, sink) case "apt_upgrade": svc := strings.TrimPrefix(targetSlug, "lxc:") cmd = fmt.Sprintf("apt update -qq 2>&1 >/dev/null && apt upgrade -y -qq 2>&1; echo '---'; systemctl is-active %s || true", svc) - output, err = sshExec(ctx, host, user, cmd) + output, err = sshExecStream(ctx, host, user, cmd, sink) case "pct_create": var cfg struct { - VMID int `json:"vmid"` - Hostname string `json:"hostname"` - Cores int `json:"cores"` - Memory int `json:"memory"` - DiskGB int `json:"disk_gb"` - IP string `json:"ip"` - GW string `json:"gw"` - Bridge string `json:"bridge"` // e.g. vmbr0/vmbr1 — which bridge actually reaches the target subnet on this host varies per host, don't assume vmbr0 - Storage string `json:"storage"` - Template string `json:"template"` - Privileged flexBool `json:"privileged"` - Nesting flexBool `json:"nesting"` - Mounts []string `json:"mounts"` - Nameserver string `json:"nameserver"` - Searchdomain string `json:"searchdomain"` + VMID int `json:"vmid"` + Hostname string `json:"hostname"` + Cores int `json:"cores"` + Memory int `json:"memory"` + DiskGB int `json:"disk_gb"` + IP string `json:"ip"` + GW string `json:"gw"` + Bridge string `json:"bridge"` // e.g. vmbr0/vmbr1 — which bridge actually reaches the target subnet on this host varies per host, don't assume vmbr0 + Storage string `json:"storage"` + Template string `json:"template"` + Privileged flexBool `json:"privileged"` + Nesting flexBool `json:"nesting"` + Mounts []string `json:"mounts"` + Nameserver string `json:"nameserver"` + Searchdomain string `json:"searchdomain"` // No services/post_install here anymore — pct_create is atomic // (create + start + register only). Installing packages and // running setup scripts is the agent's job via follow-up `run` @@ -504,7 +579,7 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, } slog.Info("httpapi: pct_create running", "vmid", cfg.VMID, "hostname", cfg.Hostname, "cmd", createCmd) - output, err = sshExec(ctx, host, user, 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 @@ -579,7 +654,7 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, return } cmd = wrap(cfg.Command) - output, err = sshExec(ctx, host, user, cmd) + output, err = sshExecStream(ctx, host, user, cmd, sink) default: slog.Error("httpapi: unknown gated action for approved execution", "action", action, "execution_id", execID) diff --git a/internal/httpapi/default_checks.go b/internal/httpapi/default_checks.go index ff6f7fa..b9509fa 100644 --- a/internal/httpapi/default_checks.go +++ b/internal/httpapi/default_checks.go @@ -4,10 +4,30 @@ import ( "context" "github.com/dtoro/oikos/internal/checkdefaults" + "github.com/dtoro/oikos/internal/db" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) -func ensureDefaultChecks(ctx context.Context, tx pgx.Tx, entityID uuid.UUID, slug, entityType string, attrsJSON []byte) { - checkdefaults.Ensure(ctx, tx, entityID, slug, entityType, attrsJSON) +// ensureDefaultChecks derives an entity's default checks from the monitoring +// kinds its type declares. +// +// Note the ordering caveat: an entity created through the API usually has no +// edges yet, so a type whose address comes from its host (a service) will +// produce no checks on this pass. That gap is real and deliberately visible — +// coverageSweep reports it, and the next inventory ingest fills it in once +// the hosting edge exists. +func ensureDefaultChecks(ctx context.Context, tx pgx.Tx, entityID uuid.UUID, slug, entityType, name string, attrsJSON []byte) error { + tree, err := db.LoadTypeTree(ctx, tx) + if err != nil { + return err + } + res, err := checkdefaults.Ensure(ctx, tx, tree, checkdefaults.Target{ + ID: entityID, Slug: slug, Type: entityType, Name: name, Attrs: attrsJSON, + }) + if err != nil { + return err + } + checkdefaults.LogResult(slug, entityType, res) + return nil } diff --git a/internal/httpapi/execution_logs.go b/internal/httpapi/execution_logs.go new file mode 100644 index 0000000..61dcdf2 --- /dev/null +++ b/internal/httpapi/execution_logs.go @@ -0,0 +1,55 @@ +package httpapi + +import ( + "encoding/json" + "net/http" + "strconv" + "strings" + + "github.com/dtoro/oikos/internal/execlog" + "github.com/go-chi/chi/v5" + "github.com/google/uuid" +) + +// serveExecutionLogs returns an execution's streamed command output. +// +// Registered as a carve-out rather than through the OpenAPI codegen for the +// same reason as /activity/recent: it is a recency-ordered projection with no +// schema type yet. Without this the execution_logs rows would be write-only — +// which is the exact shape of the bugs this whole change set has been about. +func (s *Server) serveExecutionLogs(w http.ResponseWriter, req *http.Request) { + ctx := req.Context() + + rawID := chi.URLParam(req, "id") + execID, err := uuid.Parse(rawID) + if err != nil { + writeProblem(w, req, http.StatusBadRequest, "invalid execution id", rawID) + return + } + + limit := 1000 + if l := req.URL.Query().Get("limit"); l != "" { + if n, perr := strconv.Atoi(l); perr == nil && n > 0 && n <= 5000 { + limit = n + } + } + + chunks, err := execlog.Read(ctx, s.pool, execID, limit) + if err != nil { + writeProblem(w, req, http.StatusInternalServerError, "query failed", err.Error()) + return + } + + // Also hand back the concatenation, since that is what a caller tailing + // output actually wants to render. + var combined strings.Builder + for _, c := range chunks { + combined.WriteString(c.Chunk) + } + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "items": chunks, + "combined": combined.String(), + }) +} diff --git a/internal/httpapi/executions.go b/internal/httpapi/executions.go index ae6aa22..e1a6098 100644 --- a/internal/httpapi/executions.go +++ b/internal/httpapi/executions.go @@ -4,6 +4,8 @@ import ( "context" "encoding/json" "fmt" + "strings" + "time" "github.com/dtoro/oikos/internal/db/sqlcgen" "github.com/dtoro/oikos/internal/domain" @@ -15,8 +17,22 @@ import ( // ─── Executions ──────────────────────────────────────────────────────── +// ListExecutions returns executions newest-first. +// +// The target/action/correlation_id filters are declared in the OpenAPI spec and +// generated into the request struct, but were never bound — so +// `GET /executions?target=` silently returned the first page of the whole +// fleet. Ordering was by target slug, which is neither useful for a history +// view nor unique enough to paginate on: several executions share a target, so +// a slug cursor could skip or repeat rows. func (s *Server) ListExecutions(ctx context.Context, req gen.ListExecutionsRequestObject) (gen.ListExecutionsResponseObject, error) { limit := clampLimit(req.Params.Limit) + + cursorTime, cursorID, err := parseExecutionCursor(req.Params.Cursor) + if err != nil { + return nil, err + } + rows, err := s.pool.Query(ctx, ` SELECT e.entity_id, e.classification_id::text, e.signal_entity_id::text, e.target_entity_id, e.action, e.risk_class, @@ -27,10 +43,18 @@ func (s *Server) ListExecutions(ctx context.Context, req gen.ListExecutionsReque FROM executions e JOIN entities te ON te.id = e.target_entity_id WHERE ($1::text IS NULL OR e.status = $1) - AND ($2::text IS NULL OR te.slug > $2) - ORDER BY te.slug - LIMIT $3`, - req.Params.Status, req.Params.Cursor, limit+1) + -- target accepts a slug or a uuid: the SPA passes an entity id, + -- while a human poking the API reaches for the slug. + AND ($2::text IS NULL OR te.slug = $2 OR e.target_entity_id::text = $2) + -- the run tool encodes action as "run:{json}", so match the verb too + AND ($3::text IS NULL OR e.action = $3 OR split_part(e.action, ':', 1) = $3) + AND ($4::text IS NULL OR e.correlation_id = $4) + AND ($5::timestamptz IS NULL + OR (e.created_at, e.entity_id) < ($5::timestamptz, $6::uuid)) + ORDER BY e.created_at DESC, e.entity_id DESC + LIMIT $7`, + req.Params.Status, req.Params.Target, req.Params.Action, req.Params.CorrelationId, + cursorTime, cursorID, limit+1) if err != nil { return nil, err } @@ -64,7 +88,9 @@ func (s *Server) ListExecutions(ctx context.Context, req gen.ListExecutionsReque var next *string if len(items) > limit { items = items[:limit] - next = &items[len(items)-1].Slug + last := items[len(items)-1] + cursor := formatExecutionCursor(last.CreatedAt, last.Id) + next = &cursor } if items == nil { items = []gen.Execution{} @@ -72,6 +98,32 @@ func (s *Server) ListExecutions(ctx context.Context, req gen.ListExecutionsReque return gen.ListExecutions200JSONResponse{Items: items, NextCursor: next}, nil } +// Executions are ordered by (created_at DESC, entity_id DESC), so the cursor +// has to carry both — created_at alone is not unique, and paginating on a +// non-unique key drops or repeats rows at page boundaries. +func formatExecutionCursor(createdAt time.Time, id uuid.UUID) string { + return createdAt.UTC().Format(time.RFC3339Nano) + "," + id.String() +} + +func parseExecutionCursor(cursor *string) (*time.Time, *uuid.UUID, error) { + if cursor == nil || *cursor == "" { + return nil, nil, nil + } + rawTime, rawID, ok := strings.Cut(*cursor, ",") + if !ok { + return nil, nil, domain.ErrInvalidInput + } + t, err := time.Parse(time.RFC3339Nano, rawTime) + if err != nil { + return nil, nil, domain.ErrInvalidInput + } + id, err := uuid.Parse(rawID) + if err != nil { + return nil, nil, domain.ErrInvalidInput + } + return &t, &id, nil +} + func (s *Server) GetExecution(ctx context.Context, req gen.GetExecutionRequestObject) (gen.GetExecutionResponseObject, error) { id, err := s.resolveEntityID(ctx, req.Id) if err != nil { diff --git a/internal/httpapi/executions_cursor_test.go b/internal/httpapi/executions_cursor_test.go new file mode 100644 index 0000000..5296c63 --- /dev/null +++ b/internal/httpapi/executions_cursor_test.go @@ -0,0 +1,64 @@ +package httpapi + +import ( + "testing" + "time" + + "github.com/google/uuid" +) + +// The cursor carries both created_at and entity_id because executions are +// ordered by the pair. created_at alone is not unique — several executions can +// share a millisecond — and paginating on a non-unique key silently drops or +// repeats rows at page boundaries. The previous cursor was the target slug, +// which is far less unique still: every execution against the same host shares +// it. +func TestExecutionCursorRoundTrips(t *testing.T) { + created := time.Date(2026, 7, 28, 9, 15, 30, 123456789, time.UTC) + id := uuid.MustParse("018f3a2b-0000-7000-8000-000000000042") + + cursor := formatExecutionCursor(created, id) + + gotTime, gotID, err := parseExecutionCursor(&cursor) + if err != nil { + t.Fatalf("parse: %v", err) + } + if !gotTime.Equal(created) { + t.Errorf("time round-trip: got %v, want %v", gotTime, created) + } + if *gotID != id { + t.Errorf("id round-trip: got %v, want %v", *gotID, id) + } +} + +func TestExecutionCursorNanosecondsSurvive(t *testing.T) { + // Truncating to seconds would make the cursor ambiguous for executions + // started in the same second, which is the normal case for a plan whose + // steps run back to back. + a := time.Date(2026, 7, 28, 9, 15, 30, 1, time.UTC) + b := time.Date(2026, 7, 28, 9, 15, 30, 2, time.UTC) + id := uuid.New() + + if formatExecutionCursor(a, id) == formatExecutionCursor(b, id) { + t.Error("cursors one nanosecond apart must not collide") + } +} + +func TestExecutionCursorRejectsGarbage(t *testing.T) { + empty := "" + tm, id, err := parseExecutionCursor(&empty) + if err != nil || tm != nil || id != nil { + t.Errorf("empty cursor should mean 'no cursor', got %v/%v/%v", tm, id, err) + } + + if tm, id, err := parseExecutionCursor(nil); err != nil || tm != nil || id != nil { + t.Errorf("nil cursor should mean 'no cursor', got %v/%v/%v", tm, id, err) + } + + for _, bad := range []string{"nonsense", "2026-07-28T09:15:30Z", "notatime,018f3a2b-0000-7000-8000-000000000042", "2026-07-28T09:15:30Z,notauuid"} { + b := bad + if _, _, err := parseExecutionCursor(&b); err == nil { + t.Errorf("cursor %q should have been rejected", bad) + } + } +} diff --git a/internal/httpapi/impl.go b/internal/httpapi/impl.go index 9203337..8e1257c 100644 --- a/internal/httpapi/impl.go +++ b/internal/httpapi/impl.go @@ -999,7 +999,9 @@ func (s *Server) CreateEntity(ctx context.Context, req gen.CreateEntityRequestOb return nil, eventErr } - ensureDefaultChecks(ctx, tx, inserted.ID, slug, req.Body.Type, attrsJSON) + if err := ensureDefaultChecks(ctx, tx, inserted.ID, slug, req.Body.Type, inserted.Name, attrsJSON); err != nil { + return nil, err + } if err := tx.Commit(ctx); err != nil { return nil, err @@ -1265,7 +1267,9 @@ func (s *Server) EnrollClient(ctx context.Context, req gen.EnrollClientRequestOb "info", "oikos-api", "", map[string]any{"slug": req.Body.Slug, "type": current.Type}) - ensureDefaultChecks(ctx, tx, id, req.Body.Slug, current.Type, attrsJSON) + if err := ensureDefaultChecks(ctx, tx, id, req.Body.Slug, current.Type, current.Name, attrsJSON); err != nil { + return nil, err + } if err := tx.Commit(ctx); err != nil { return nil, err diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index 9565a83..08b53ef 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -111,6 +111,7 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Hand // /api/v1/knowledge/content/{id} — returns raw markdown, not a gen type // /api/v1/activity/recent — recency-ordered, not paginated // /api/v1/activity/session/{id} — session-scoped aggregation + // /api/v1/executions/{id}/logs — streamed command output, no schema type // /api/v1/learning/timeline — derived view, no backing schema type // /api/v1/learning/trend — derived view, no backing schema type // @@ -207,6 +208,11 @@ func NewHandler(ctx context.Context, pool *db.Pool, cfg config.Config) http.Hand // per-session "what did this session do" digest. // (See "Non-OpenAPI routes" carve-out block above.) r.With(combinedAuth(cfg, false)).Get("/api/v1/activity/recent", s.serveRecentActivity) + // Streamed command output for one execution — a projection over + // execution_logs with no schema type yet (same carve-out rationale as + // /activity/recent above). Nests cleanly under the generated + // /executions/{id} subtree: chi accepts sibling children on a param node. + r.With(combinedAuth(cfg, false)).Get("/api/v1/executions/{id}/logs", s.serveExecutionLogs) r.With(combinedAuth(cfg, false)).Get("/api/v1/activity/session/{id}", s.serveSessionDigest) // Learning view: capability timeline + success trend, both derived from @@ -349,10 +355,10 @@ func staticTokenActor(cfg config.Config, raw string) (actor, bool) { // jwtVerificationKey holds a parsed RSA public key or HMAC secret for JWT // verification, identified by its key ID (kid). type jwtVerificationKey struct { - Kid string - Alg string - Key any // *rsa.PublicKey or []byte for HMAC - IsHMAC bool + Kid string + Alg string + Key any // *rsa.PublicKey or []byte for HMAC + IsHMAC bool } // discoverJWKSURI fetches the OIDC discovery document and extracts the @@ -598,8 +604,8 @@ func resolveOIDCTokenURL(issuer string) string { func (s *Server) serveOIDCConfig(w http.ResponseWriter, _ *http.Request, cfg config.Config) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]string{ - "issuer": cfg.OIDCIssuer, - "client_id": cfg.OIDCClientID, + "issuer": cfg.OIDCIssuer, + "client_id": cfg.OIDCClientID, "authorization_endpoint": resolveOIDCEndpointURL(cfg.OIDCIssuer, "/authorize/"), }) } @@ -863,4 +869,4 @@ func ListenAndServe(ctx context.Context, pool *db.Pool, cfg config.Config) error defer cancel() return srv.Shutdown(shutdownCtx) } -} \ No newline at end of file +} diff --git a/internal/mcp/server.go b/internal/mcp/server.go index f25ffaf..484a8a6 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -22,6 +22,7 @@ import ( "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/db/sqlcgen" + "github.com/dtoro/oikos/internal/execlog" "github.com/dtoro/oikos/internal/observability" "github.com/dtoro/oikos/internal/policy" "github.com/google/jsonschema-go/jsonschema" @@ -329,7 +330,37 @@ func initSSH() { // goroutine forever with no way for the caller to ever get an answer. const sshExecTimeout = 10 * time.Minute +// streamWriter buffers everything it is given while forwarding each write to a +// sink. Assigning one to session.Stdout and another (sharing the same buffer) +// to session.Stderr reproduces CombinedOutput's interleaving exactly, in the +// order the remote end actually produced it — which reading from StdoutPipe +// and StderrPipe separately would not guarantee. +type streamWriter struct { + mu *sync.Mutex + buf *bytes.Buffer + stream string + sink execlog.Sink +} + +func (w *streamWriter) Write(p []byte) (int, error) { + w.mu.Lock() + w.buf.Write(p) + w.mu.Unlock() + if w.sink != nil { + // Copy: the ssh library reuses p after Write returns, and the sink + // hands the bytes to a DB call that may outlive this frame. + w.sink(w.stream, append([]byte(nil), p...)) + } + return len(p), nil +} + 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) { initSSH() if len(sshKey) == 0 { return "", fmt.Errorf("no SSH key available") @@ -363,16 +394,28 @@ func sshExec(ctx context.Context, host, user, command string) (string, error) { } defer session.Close() - type result struct { - out []byte - err error + var ( + mu sync.Mutex + buf bytes.Buffer + ) + session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink} + session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink} + + // collected returns whatever output has arrived so far. Callable while the + // command is still running, which is what makes partial output on timeout + // possible. + collected := func() string { + mu.Lock() + defer mu.Unlock() + return strings.TrimSpace(buf.String()) } - done := make(chan result, 1) + + done := make(chan error, 1) go func() { - // Recovers a panic in CombinedOutput (SSH library internals, rare but - // not impossible) and reports it as a failed command instead of - // crashing the whole api process — every gated action runs through - // this function, so an unrecovered panic here would take down every + // Recovers a panic in the SSH library internals (rare but not + // impossible) and reports it as a failed command instead of crashing + // the whole api process — every gated action runs through this + // function, so an unrecovered panic here would take down every // concurrently-running task's execution, not just this one. Without // this, a panic would ALSO silently degrade to "wait out the full // timeout" (done never receives, the select below falls through to @@ -381,36 +424,40 @@ func sshExec(ctx context.Context, host, user, command string) (string, error) { // finds out now, not after sshExecTimeout. defer func() { if r := recover(); r != nil { - done <- result{nil, fmt.Errorf("panic in ssh exec: %v", r)} + done <- fmt.Errorf("panic in ssh exec: %v", r) } }() - out, err := session.CombinedOutput(command) - done <- result{out, err} + // Run rather than CombinedOutput so the assigned writers are used; + // Run returns only after both streams have been fully drained. + done <- session.Run(command) }() select { - case r := <-done: - text := strings.TrimSpace(string(r.out)) + case err := <-done: + text := collected() // A non-zero exit MUST surface as an error — matching the fix // applied to httpapi's sshExec (this copy still had the original // bug: only erroring when there was no output at all, so a command // that failed but printed something was silently reported as // success). - if r.err != nil { + if err != nil { if text != "" { - return text, fmt.Errorf("%w: %s", r.err, text) + return text, fmt.Errorf("%w: %s", err, text) } - return text, fmt.Errorf("exec: %w", r.err) + return text, fmt.Errorf("exec: %w", err) } return text, nil case <-time.After(sshExecTimeout): session.Close() client.Close() - return "", fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host) + // Return what the command managed to print before it hung. This used + // to return "", discarding everything — so a hung command, the case + // where the output matters most, was the one case that left no trace. + return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host) case <-ctx.Done(): session.Close() client.Close() - return "", ctx.Err() + return collected(), ctx.Err() } } @@ -631,6 +678,56 @@ func resolveProxmoxHostSlug(ctx context.Context, pool *db.Pool, entitySlug, host // fleet's reverse proxy) executed instantly with no approval at all. Routing // every mutating path through the same classifier + approval-queue logic // closes that gap without special-casing each caller. +// autoRun resolves a target, runs the command, and finalizes the execution +// with full timing. +// +// The three auto-run windows (read-only, assent, destructive) each carried +// their own copy of this logic, and none of them wrote duration_ms, started_at +// or completed_at — so every auto-run execution landed in the ledger with no +// timing at all, and the Ops "Duration" column was empty for exactly the +// executions that run most often. +func autoRun(ctx context.Context, pool *db.Pool, id uuid.UUID, targetSlug, command string) (string, error) { + startedAt := time.Now() + if _, err := pool.Exec(ctx, + `UPDATE executions SET status='running', started_at=$2 WHERE entity_id=$1`, + id, startedAt); err != nil { + slog.Error("mcp: mark execution running", "error", err, "execution_id", id) + } + + finalize := func(status string, result []byte) { + if _, err := pool.Exec(ctx, + `UPDATE executions SET status=$2, result=$3::jsonb, duration_ms=$4, + started_at=$5, completed_at=now() + WHERE entity_id=$1`, + id, status, result, int(time.Since(startedAt).Milliseconds()), startedAt); err != nil { + slog.Error("mcp: finalize execution", "error", err, "execution_id", id) + } + } + + host, user, wrap, err := resolveExecTarget(ctx, pool, targetSlug) + if err != nil { + finalize("failed", jsonErr("%s", err.Error())) + return "", err + } + + var correlationID string + if qerr := pool.QueryRow(ctx, + `SELECT correlation_id FROM executions WHERE entity_id = $1`, id).Scan(&correlationID); qerr != nil { + correlationID = "" + } + sink, flush := execlog.New(ctx, pool, id, correlationID) + + out, err := sshExecStream(ctx, host, user, wrap(command), sink) + flush() + if err != nil { + finalize("failed", jsonErr("%s: %s", err.Error(), out)) + return out, err + } + + finalize("completed", jsonOut(out)) + return out, nil +} + func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.UUID, targetSlug, command, purpose, declaredRisk, sessionID string) *mcp.CallToolResult { riskClass := policy.ClassifyCommand(command, declaredRisk) runParams, _ := json.Marshal(map[string]string{"command": command, "purpose": purpose}) @@ -686,7 +783,20 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid. } id, _ := uuid.NewV7() - correlationID := uuid.New().String() + // Correlate the execution to the chat session that asked for it. This was + // a fresh random UUID per execution, which correlated nothing — every row + // had a unique value, so the correlation_id column and the + // ?correlation_id= filter could only ever match one execution. + // + // Using the session id makes the field mean what it says ("what did this + // session do?") and is what lets the chat tail live output: execution + // events carry correlation_id, so the UI can match them to the session on + // screen without a lookup. Falls back to a random id when there is no + // session to scope to, keeping the column non-empty. + correlationID := sessionID + if correlationID == "" || correlationID == "ephemeral" { + correlationID = uuid.New().String() + } execName := "run on " + targetSlug + " (" + id.String() + ")" execSlug := "exec:" + targetSlug + ":" + id.String() if _, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`, @@ -713,19 +823,29 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid. id, "task:"+sessionID) } - if riskClass == policy.RiskReadOnly { - host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug) - if rerr != nil { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error())) - return textResult(fmt.Sprintf("resolve target: %v", rerr)) - } - out, xerr := sshExec(ctx, host, user, wrap(command)) + // read_only and reversible_low both run unattended, as seeds/policy.yaml + // and .agents/OIKOS.md declare ("reversible_low — restart, cache clear, + // sync pull. Unattended + ledger."). + // + // reversible_low had no branch here, so it fell through to the gate. That + // looked stricter but was actually perverse: computeCommandRisk never + // returns reversible_low — the class can ONLY arise when the agent + // declares it on a command the classifier already scored read_only + // (ClassifyCommand keeps the higher of the two). So an agent that + // honestly flagged "this restarts something" got gated, while the same + // command with no declaration auto-ran. That penalised candor and gave + // the agent a reason to stay quiet. + // + // Auto-running it is no more permissive than the read_only branch above, + // because read_only is the only computed class it can accompany. An + // agent still cannot talk a command DOWN: declaring reversible_low on + // something computed as config_mutation keeps config_mutation. + if riskClass == policy.RiskReadOnly || riskClass == policy.RiskReversibleLow { + out, xerr := autoRun(ctx, pool, id, targetSlug, command) if xerr != nil { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out)) return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out)) } - pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out)) - return textResult(fmt.Sprintf("run on %s (read_only, auto): %s", targetSlug, out)) + return textResult(fmt.Sprintf("run on %s (%s, auto): %s", targetSlug, riskClass, out)) } // Assent window: if the operator recently approved a plan in this @@ -739,17 +859,10 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid. // consent. The assent window, opened only on operator approval, is the // sole gate for config_mutation auto-run.) if riskClass == policy.RiskConfigMutation && assentWindowActive(ctx, pool, agentID, sessionID) { - host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug) - if rerr != nil { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error())) - return textResult(fmt.Sprintf("resolve target: %v", rerr)) - } - out, xerr := sshExec(ctx, host, user, wrap(command)) + out, xerr := autoRun(ctx, pool, id, targetSlug, command) if xerr != nil { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out)) return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out)) } - pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out)) slog.Info("mcp: run auto-executed via assent window", "target", targetSlug, "execution_id", id) return textResult(fmt.Sprintf("run on %s (config_mutation, auto via assent window): %s", targetSlug, out)) } @@ -761,17 +874,10 @@ func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid. // operator isn't asked to re-type "I confirm" for every single command // against the thing they just confirmed. if riskClass == policy.RiskDestructive && destructiveWindowActive(ctx, pool, agentID, targetSlug, sessionID) { - host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug) - if rerr != nil { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error())) - return textResult(fmt.Sprintf("resolve target: %v", rerr)) - } - out, xerr := sshExec(ctx, host, user, wrap(command)) + out, xerr := autoRun(ctx, pool, id, targetSlug, command) if xerr != nil { - pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out)) return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out)) } - pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out)) slog.Info("mcp: run auto-executed via destructive window", "target", targetSlug, "execution_id", id) return textResult(fmt.Sprintf("run on %s (destructive, auto via confirmed-target window): %s", targetSlug, out)) } diff --git a/internal/mcp/sshexec_test.go b/internal/mcp/sshexec_test.go new file mode 100644 index 0000000..c19e124 --- /dev/null +++ b/internal/mcp/sshexec_test.go @@ -0,0 +1,159 @@ +package mcp + +// Streaming tests for sshExecStream against a real SSH endpoint. Guarded by +// OIKOS_SSH_TEST_HOST — skipped when unset. Run with: +// +// OIKOS_SSH_TEST_HOST=localhost OIKOS_SSH_USER=$USER \ +// OIKOS_SSH_KEY_PATH=~/.ssh/id_ed25519 go test ./internal/mcp/ -run TestSSHExecStream +// +// These matter because the whole point of the change is behaviour that only +// appears over time: that output arrives *before* the command exits, and that +// a command killed mid-flight still leaves what it printed. + +import ( + "context" + "os" + "strings" + "sync" + "testing" + "time" +) + +func sshTestHost(t *testing.T) string { + t.Helper() + host := os.Getenv("OIKOS_SSH_TEST_HOST") + if host == "" { + t.Skip("OIKOS_SSH_TEST_HOST not set — skipping live SSH test") + } + return host +} + +// The core claim: chunks reach the sink while the command is still running, +// not in one lump at the end. A command that prints, sleeps, then prints must +// deliver its first chunk well before it exits. +func TestSSHExecStreamDeliversOutputBeforeExit(t *testing.T) { + host := sshTestHost(t) + + var ( + mu sync.Mutex + chunks []string + firstA time.Time + ) + sink := func(stream string, chunk []byte) { + mu.Lock() + defer mu.Unlock() + if firstA.IsZero() { + firstA = time.Now() + } + chunks = append(chunks, string(chunk)) + } + + start := time.Now() + out, err := sshExecStream(context.Background(), host, os.Getenv("OIKOS_SSH_USER"), + "echo FIRST; sleep 2; echo SECOND", sink) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("sshExecStream: %v (out=%q)", err, out) + } + + mu.Lock() + joined := strings.Join(chunks, "") + firstAt := firstA.Sub(start) + mu.Unlock() + + if !strings.Contains(out, "FIRST") || !strings.Contains(out, "SECOND") { + t.Errorf("combined output lost content: %q", out) + } + if !strings.Contains(joined, "FIRST") || !strings.Contains(joined, "SECOND") { + t.Errorf("sink did not receive the full output: %q", joined) + } + if elapsed < 2*time.Second { + t.Fatalf("command returned in %v — the sleep did not run, test is not measuring what it claims", elapsed) + } + // The first chunk must land near the start, not at the end. + if firstAt > elapsed/2 { + t.Errorf("first chunk arrived after %v of a %v command — output is still being buffered to the end", + firstAt, elapsed) + } +} + +// stderr must reach the sink too, and land in the combined output, matching +// what CombinedOutput used to return. +func TestSSHExecStreamCapturesBothStreams(t *testing.T) { + host := sshTestHost(t) + + var mu sync.Mutex + seen := map[string]bool{} + sink := func(stream string, chunk []byte) { + mu.Lock() + defer mu.Unlock() + seen[stream] = true + } + + out, err := sshExecStream(context.Background(), host, os.Getenv("OIKOS_SSH_USER"), + "echo TO_STDOUT; echo TO_STDERR 1>&2", sink) + if err != nil { + t.Fatalf("sshExecStream: %v (out=%q)", err, out) + } + + if !strings.Contains(out, "TO_STDOUT") || !strings.Contains(out, "TO_STDERR") { + t.Errorf("combined output missing a stream: %q", out) + } + mu.Lock() + defer mu.Unlock() + if !seen["stdout"] { + t.Error("sink never saw a stdout chunk") + } + if !seen["stderr"] { + t.Error("sink never saw a stderr chunk") + } +} + +// A cancelled command used to return "" — everything it had printed was +// thrown away. The hung case is exactly when that output is worth having. +func TestSSHExecStreamKeepsPartialOutputOnCancel(t *testing.T) { + host := sshTestHost(t) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + out, err := sshExecStream(ctx, host, os.Getenv("OIKOS_SSH_USER"), + "echo BEFORE_HANG; sleep 30; echo NEVER", nil) + + if err == nil { + t.Fatal("expected a context error for a command that outlives the deadline") + } + if !strings.Contains(out, "BEFORE_HANG") { + t.Errorf("partial output was discarded on cancel: %q", out) + } + if strings.Contains(out, "NEVER") { + t.Errorf("command should not have completed: %q", out) + } +} + +// A nil sink must behave exactly as the old CombinedOutput path did. +func TestSSHExecNilSinkStillReturnsOutput(t *testing.T) { + host := sshTestHost(t) + + out, err := sshExec(context.Background(), host, os.Getenv("OIKOS_SSH_USER"), "echo PLAIN") + if err != nil { + t.Fatalf("sshExec: %v", err) + } + if out != "PLAIN" { + t.Errorf("out = %q, want %q (output is trimmed)", out, "PLAIN") + } +} + +// A non-zero exit must surface as an error while still returning the output. +func TestSSHExecStreamNonZeroExitIsAnError(t *testing.T) { + host := sshTestHost(t) + + out, err := sshExecStream(context.Background(), host, os.Getenv("OIKOS_SSH_USER"), + "echo PRINTED_THEN_FAILED; exit 3", nil) + if err == nil { + t.Fatal("a non-zero exit that printed output must still be an error") + } + if !strings.Contains(out, "PRINTED_THEN_FAILED") { + t.Errorf("output lost on failure: %q", out) + } +} diff --git a/internal/ontology/monitoring.go b/internal/ontology/monitoring.go new file mode 100644 index 0000000..5146f54 --- /dev/null +++ b/internal/ontology/monitoring.go @@ -0,0 +1,90 @@ +package ontology + +// Monitoring resolution: which check kinds an entity type warrants. +// +// Coverage is not uniform. A `service` warrants an HTTP probe; a `site` is a +// physical location with nothing to probe; a `dns-zone` warrants a check whose +// checker does not exist yet. Collapsing those three into "has no check_def" +// is what made the fleet's monitoring gap invisible — 86 of 89 active entities +// had no check, and staleSweep's INNER JOIN against check_defs meant none of +// them could ever be marked stale. +// +// So the declaration lives on the entity TYPE, in seeds/ontology.yaml, and +// resolves through the same is-a hierarchy the validator already walks: +// declaring `monitoring: [ping, resource]` on abstract `machine` covers +// proxmox-host, standalone-server, workstation and appliance. + +// MonitoringResolution is the outcome of resolving a type's monitoring +// declaration. The three states are deliberately distinguishable: +// +// Declared=false — nobody in the chain said anything. An ontology +// gap: report it, but as a modelling problem +// rather than as a fleet monitoring problem. +// Declared=true, len(0) — explicitly unmonitorable. Working as intended; +// never raise an `unmonitored` signal for it. +// Declared=true, len(n) — these kinds are expected to exist. +type MonitoringResolution struct { + Kinds []string + + // Declared reports whether anything in the chain (or the layer default) + // settled the question. + Declared bool + + // Source names the type that supplied the answer — the type itself, an + // ancestor, or "" when the layer default applied. Useful in log lines + // that explain why an entity has the checks it has. + Source string +} + +// None reports an explicit "this type is not monitored". +func (m MonitoringResolution) None() bool { + return m.Declared && len(m.Kinds) == 0 +} + +// Wants reports whether the type expects a check of this kind. +func (m MonitoringResolution) Wants(kind string) bool { + for _, k := range m.Kinds { + if k == kind { + return true + } + } + return false +} + +// Monitoring resolves the check kinds a type warrants, walking parent types +// until one carries a declaration. +// +// Types outside the infrastructure layer (governance, cognition, meta) fall +// back to an implicit "none": a signal, an approval, a document and a person +// are records, not running things. That default keeps ~20 record types out of +// the ontology without needing an explicit `monitoring: none` on each, while +// still treating an undeclared *infrastructure* type as a genuine gap — those +// are the ones somebody should have made a decision about. A non-infrastructure +// type that really is probeable (agent, which serves a gateway on :8092) just +// declares its kinds explicitly and wins on the first rule. +func (t *TypeTree) Monitoring(typ string) MonitoringResolution { + seen := map[string]bool{} + for cur := typ; cur != ""; cur = t.Types[cur].Parent { + info, ok := t.Types[cur] + if !ok { + break + } + if seen[cur] { + break // cycle guard — ingest rejects cycles, belt and braces + } + seen[cur] = true + + if info.Monitoring != nil { + return MonitoringResolution{ + Kinds: *info.Monitoring, + Declared: true, + Source: cur, + } + } + } + + if info, ok := t.Types[typ]; ok && info.Layer != "infrastructure" { + return MonitoringResolution{Declared: true} + } + return MonitoringResolution{} +} diff --git a/internal/ontology/monitoring_test.go b/internal/ontology/monitoring_test.go new file mode 100644 index 0000000..adf8c6c --- /dev/null +++ b/internal/ontology/monitoring_test.go @@ -0,0 +1,121 @@ +package ontology + +import ( + "testing" +) + +func kinds(v ...string) *[]string { + s := append([]string{}, v...) + return &s +} + +// monitoringTree mirrors the real shape of seeds/ontology.yaml: a declaration +// on an abstract type that concrete subtypes inherit, an explicit none on a +// topological type, a probeable type outside the infrastructure layer, and an +// undeclared infrastructure type (the ontology gap this is meant to catch). +func monitoringTree() *TypeTree { + return &TypeTree{ + Types: map[string]TypeInfo{ + "entity": {IsAbstract: true, Layer: "meta"}, + "compute-entity": {Parent: "entity", IsAbstract: true, Layer: "infrastructure"}, + "machine": {Parent: "compute-entity", IsAbstract: true, Layer: "infrastructure", + Monitoring: kinds("ping", "resource")}, + "proxmox-host": {Parent: "machine", Layer: "infrastructure"}, + "workstation": {Parent: "machine", Layer: "infrastructure"}, + "service": {Parent: "entity", Layer: "infrastructure", Monitoring: kinds("http", "process")}, + "site": {Parent: "entity", Layer: "infrastructure", Monitoring: kinds()}, + "vlan": {Parent: "entity", Layer: "infrastructure"}, // undeclared: a gap + "agent": {Parent: "entity", Layer: "governance", Monitoring: kinds("http")}, + "signal": {Parent: "entity", Layer: "cognition"}, + "document": {Parent: "entity", Layer: "governance"}, + }, + } +} + +func TestMonitoringResolvesThroughHierarchy(t *testing.T) { + tree := monitoringTree() + + cases := []struct { + typ string + wantKinds []string + wantDecl bool + wantSource string + desc string + }{ + {"machine", []string{"ping", "resource"}, true, "machine", "declared on itself"}, + {"proxmox-host", []string{"ping", "resource"}, true, "machine", "inherited from abstract parent"}, + {"workstation", []string{"ping", "resource"}, true, "machine", "inherited by a sibling too"}, + {"service", []string{"http", "process"}, true, "service", "declared on itself"}, + {"site", nil, true, "site", "explicitly none — not a gap"}, + {"agent", []string{"http"}, true, "agent", "explicit declaration beats the layer default"}, + {"signal", nil, true, "", "cognition layer is implicitly none"}, + {"document", nil, true, "", "governance layer is implicitly none"}, + {"vlan", nil, false, "", "undeclared infrastructure type is a genuine gap"}, + {"nonexistent", nil, false, "", "unknown type resolves to undeclared"}, + } + + for _, c := range cases { + got := tree.Monitoring(c.typ) + if got.Declared != c.wantDecl { + t.Errorf("%s (%s): Declared = %v, want %v", c.typ, c.desc, got.Declared, c.wantDecl) + } + if got.Source != c.wantSource { + t.Errorf("%s (%s): Source = %q, want %q", c.typ, c.desc, got.Source, c.wantSource) + } + if len(got.Kinds) != len(c.wantKinds) { + t.Errorf("%s (%s): Kinds = %v, want %v", c.typ, c.desc, got.Kinds, c.wantKinds) + continue + } + for i, k := range c.wantKinds { + if got.Kinds[i] != k { + t.Errorf("%s (%s): Kinds[%d] = %q, want %q", c.typ, c.desc, i, got.Kinds[i], k) + } + } + } +} + +// The distinction between these two is what keeps coverageSweep from raising +// permanent, unresolvable signals against entities that are working as intended. +func TestMonitoringNoneIsNotTheSameAsUndeclared(t *testing.T) { + tree := monitoringTree() + + site := tree.Monitoring("site") + if !site.None() { + t.Error("site declared `monitoring: none`, expected None() to report true") + } + + vlan := tree.Monitoring("vlan") + if vlan.None() { + t.Error("vlan declared nothing at all — None() must not claim it opted out") + } + if vlan.Declared { + t.Error("vlan is an undeclared infrastructure type; it should read as a gap") + } +} + +func TestMonitoringWants(t *testing.T) { + tree := monitoringTree() + + svc := tree.Monitoring("service") + if !svc.Wants("http") { + t.Error("service should want an http check") + } + if svc.Wants("resource") { + t.Error("service should not want a resource check") + } + if tree.Monitoring("site").Wants("http") { + t.Error("an explicitly unmonitorable type wants nothing") + } +} + +func TestMonitoringSurvivesParentCycle(t *testing.T) { + // The ingest rejects cycles; this guards the walker regardless. + tree := &TypeTree{Types: map[string]TypeInfo{ + "a": {Parent: "b", Layer: "infrastructure"}, + "b": {Parent: "a", Layer: "infrastructure"}, + }} + got := tree.Monitoring("a") + if got.Declared { + t.Errorf("cyclic chain declared nothing, got %+v", got) + } +} diff --git a/internal/ontology/validate.go b/internal/ontology/validate.go index 64bb0f1..963641a 100644 --- a/internal/ontology/validate.go +++ b/internal/ontology/validate.go @@ -20,6 +20,13 @@ type TypeInfo struct { Parent string IsAbstract bool LifecycleID string + Layer string + + // Monitoring is this type's own `monitoring:` declaration, or nil if it + // declared nothing (in which case the answer comes from an ancestor, or + // from the layer default). A non-nil pointer to an empty slice means + // "explicitly unmonitorable" — see TypeTree.Monitoring. + Monitoring *[]string } // RelTypeInfo is the subset of a relationship type the validator needs. diff --git a/internal/policy/command_test.go b/internal/policy/command_test.go index f69fa08..3c7f4d6 100644 --- a/internal/policy/command_test.go +++ b/internal/policy/command_test.go @@ -185,3 +185,32 @@ func TestClassifyCommand_EmptyCommand(t *testing.T) { t.Errorf("empty command should default to config_mutation (escalate), got %q", got) } } + +// reversible_low is never computed from the command text — it can only arrive +// as a declaration. These pin down the asymmetry that makes auto-running it +// safe: a declaration may raise the class but never lower it, so the only +// computed class reversible_low can accompany is read_only. +func TestReversibleLowOnlyArrivesAsADeclaration(t *testing.T) { + // Nothing in the command text alone yields reversible_low. + for _, cmd := range []string{ + "systemctl restart nginx", "uptime", "cat /etc/os-release", + "apt-get update", "docker restart web", "rm -rf /tmp/x", + } { + if got := ClassifyCommand(cmd, ""); got == RiskReversibleLow { + t.Errorf("ClassifyCommand(%q, \"\") = reversible_low; the classifier should never compute it", cmd) + } + } + + // Declaring it on a read-only command raises to reversible_low... + if got := ClassifyCommand("uptime", RiskReversibleLow); got != RiskReversibleLow { + t.Errorf("declared reversible_low over a read_only command = %q, want reversible_low", got) + } + + // ...but declaring it can never talk a riskier command down. + if got := ClassifyCommand("apt-get upgrade -y", RiskReversibleLow); got == RiskReversibleLow { + t.Error("declaring reversible_low must not lower a config_mutation command") + } + if got := ClassifyCommand("rm -rf /var/lib/x", RiskReversibleLow); got != RiskDestructive { + t.Errorf("declaring reversible_low over a destructive command = %q, want destructive", got) + } +} diff --git a/internal/scheduler/backup.go b/internal/scheduler/backup.go new file mode 100644 index 0000000..958ecd5 --- /dev/null +++ b/internal/scheduler/backup.go @@ -0,0 +1,116 @@ +package scheduler + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "time" + + "github.com/dtoro/oikos/internal/db/sqlcgen" +) + +// checkBackupFreshness reports whether a backup target has a recent artifact. +// +// The ontology has carried a `backup-target` type and a `backs-up-to` edge +// since the first seed, but nothing ever verified that a backup actually +// happened — a silent backup failure looked exactly like a working one. This +// makes staleness a Signal like any other, so it flows through the existing +// dedup, auto-resolve and notifier path rather than needing its own machinery. +// +// Config: {"path": "/opt/oikos/backups", "max_age_s": 86400, "host": …} +// +// Deliberately uses `find -mmin` rather than `-printf '%T@'` or `stat`: +// -printf is GNU-only and stat's format flag differs between GNU (-c) and BSD +// (-f). The first real target for this check is the pre-deploy pg_dump on the +// mac-mini, which is macOS — so a GNU-only probe would have silently reported +// "unknown" on the one target that motivated the check. +func checkBackupFreshness(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult { + cfg := struct { + Path string `json:"path"` + MaxAgeS int `json:"max_age_s"` + Host string `json:"host"` + User string `json:"user"` + Port int `json:"port"` + }{ + MaxAgeS: 86400, // a daily backup that has not run in 24h is stale + } + if len(cd.Config) > 0 { + _ = json.Unmarshal(cd.Config, &cfg) + } + if cfg.Path == "" || cfg.Host == "" { + return checkResult{health: "unknown", signalKind: "backup-misconfigured", + evidence: "backup check needs both a path and a host"} + } + if cfg.Port == 0 { + cfg.Port = 22 + } + if cfg.User == "" { + cfg.User = sshUser + } + if cfg.MaxAgeS <= 0 { + cfg.MaxAgeS = 86400 + } + + timeout := time.Duration(cd.TimeoutS) * time.Second + if timeout <= 0 { + timeout = 30 * time.Second + } + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + minutes := cfg.MaxAgeS / 60 + if minutes < 1 { + minutes = 1 + } + quoted := shellSingleQuote(cfg.Path) + + // Two questions in one round trip: is there anything at all, and is any of + // it recent? "no backups ever" and "backups stopped" are different + // failures and deserve different severities. + cmd := fmt.Sprintf( + `if [ ! -d %s ]; then echo missing; else `+ + `f=$(find %s -type f -mmin -%d 2>/dev/null | head -1); `+ + `a=$(find %s -type f 2>/dev/null | head -1); `+ + `if [ -n "$f" ]; then echo fresh; elif [ -n "$a" ]; then echo stale; else echo empty; fi; fi`, + quoted, quoted, minutes, quoted) + + out, err := sshExec(ctx, cfg.Host, strconv.Itoa(cfg.Port), cfg.User, cmd, timeout) + if err != nil { + return checkResult{ + health: "unknown", signalKind: "backup-unreachable", + evidence: fmt.Sprintf("ssh %s: %v", cfg.Host, err), + err: err, + } + } + + age := time.Duration(cfg.MaxAgeS) * time.Second + switch strings.TrimSpace(string(out)) { + case "fresh": + return checkResult{health: "healthy"} + case "stale": + return checkResult{ + health: "degraded", signalKind: "backup-stale", + evidence: fmt.Sprintf("no backup in %s under %s on %s", age, cfg.Path, cfg.Host), + } + case "empty": + return checkResult{ + health: "down", signalKind: "backup-missing", + evidence: fmt.Sprintf("%s on %s exists but contains no files", cfg.Path, cfg.Host), + } + case "missing": + return checkResult{ + health: "down", signalKind: "backup-missing", + evidence: fmt.Sprintf("backup directory %s does not exist on %s", cfg.Path, cfg.Host), + } + } + return checkResult{health: "unknown", signalKind: "backup-unreachable", + evidence: fmt.Sprintf("unexpected probe output: %q", strings.TrimSpace(string(out)))} +} + +// shellSingleQuote makes a path safe to embed in the remote sh command. Paths +// come from check_defs config, which an operator or the agent can write. +func shellSingleQuote(s string) string { + return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" +} diff --git a/internal/scheduler/backup_test.go b/internal/scheduler/backup_test.go new file mode 100644 index 0000000..2e1ad14 --- /dev/null +++ b/internal/scheduler/backup_test.go @@ -0,0 +1,101 @@ +package scheduler + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/dtoro/oikos/internal/db/sqlcgen" +) + +func backupCheckDef(t *testing.T, config string) sqlcgen.ListEnabledCheckDefsRow { + t.Helper() + return sqlcgen.ListEnabledCheckDefsRow{ + Kind: "backup-freshness", + Config: []byte(config), + TimeoutS: 15, + } +} + +// A misconfigured check must say so rather than quietly reporting healthy — +// "no path configured" and "backup ran fine" must never look the same. +func TestBackupFreshnessRejectsIncompleteConfig(t *testing.T) { + for _, c := range []struct{ desc, config string }{ + {"no path", `{"host":"localhost"}`}, + {"no host", `{"path":"/tmp"}`}, + {"empty", `{}`}, + } { + got := checkBackupFreshness(context.Background(), backupCheckDef(t, c.config)) + if got.health != "unknown" || got.signalKind != "backup-misconfigured" { + t.Errorf("%s: got health=%q kind=%q, want unknown/backup-misconfigured", + c.desc, got.health, got.signalKind) + } + } +} + +// Live probe against a real SSH endpoint. Guarded by OIKOS_SSH_TEST_HOST: +// +// OIKOS_SSH_TEST_HOST=localhost OIKOS_SSH_USER=$USER \ +// OIKOS_SSH_KEY_PATH=~/.ssh/id_ed25519 go test ./internal/scheduler/ -run TestBackupFreshnessLive +// +// The probe shell has to work on both GNU and BSD find — the first real target +// is the pre-deploy pg_dump on the macOS mac-mini, so a GNU-only construct +// would fail exactly where it matters. +func TestBackupFreshnessLiveDistinguishesTheFourStates(t *testing.T) { + host := os.Getenv("OIKOS_SSH_TEST_HOST") + if host == "" { + t.Skip("OIKOS_SSH_TEST_HOST not set — skipping live backup probe") + } + sshKeyPath = os.Getenv("OIKOS_SSH_KEY_PATH") + sshUser = os.Getenv("OIKOS_SSH_USER") + + dir := t.TempDir() + fresh := filepath.Join(dir, "fresh") + stale := filepath.Join(dir, "stale") + empty := filepath.Join(dir, "empty") + for _, d := range []string{fresh, stale, empty} { + if err := os.Mkdir(d, 0o755); err != nil { + t.Fatal(err) + } + } + if err := os.WriteFile(filepath.Join(fresh, "dump.sql"), []byte("x"), 0o644); err != nil { + t.Fatal(err) + } + oldFile := filepath.Join(stale, "dump.sql") + if err := os.WriteFile(oldFile, []byte("x"), 0o644); err != nil { + t.Fatal(err) + } + old := time.Now().Add(-72 * time.Hour) + if err := os.Chtimes(oldFile, old, old); err != nil { + t.Fatal(err) + } + + cases := []struct { + desc, path, wantHealth, wantKind string + }{ + {"recent artifact", fresh, "healthy", ""}, + {"artifact older than max_age", stale, "degraded", "backup-stale"}, + {"directory exists but is empty", empty, "down", "backup-missing"}, + {"directory does not exist", filepath.Join(dir, "nope"), "down", "backup-missing"}, + } + + for _, c := range cases { + cfg := `{"host":"` + host + `","path":"` + c.path + `","max_age_s":86400}` + got := checkBackupFreshness(context.Background(), backupCheckDef(t, cfg)) + if got.health != c.wantHealth || got.signalKind != c.wantKind { + t.Errorf("%s: got health=%q kind=%q evidence=%q, want %q/%q", + c.desc, got.health, got.signalKind, got.evidence, c.wantHealth, c.wantKind) + } + } +} + +// A path with a quote in it must not break out of the remote sh command. +func TestShellSingleQuoteEscapes(t *testing.T) { + got := shellSingleQuote(`/tmp/it's; rm -rf /`) + want := `'/tmp/it'\''s; rm -rf /'` + if got != want { + t.Errorf("shellSingleQuote = %s, want %s", got, want) + } +} diff --git a/internal/scheduler/coverage.go b/internal/scheduler/coverage.go new file mode 100644 index 0000000..ae07051 --- /dev/null +++ b/internal/scheduler/coverage.go @@ -0,0 +1,191 @@ +package scheduler + +import ( + "context" + "fmt" + "log/slog" + + "github.com/dtoro/oikos/internal/db" + "github.com/google/uuid" +) + +// UnmonitoredKind is the signal kind raised for an entity whose type declares +// monitoring it does not have. +const UnmonitoredKind = "unmonitored" + +// coverageSweep reports entities that should be monitored and are not. +// +// staleSweep can only protect an entity that already has a check — it INNER +// JOINs check_defs, so an entity with none is structurally invisible to it and +// keeps reporting its last-known health forever. This sweep covers the other +// half: it notices the absence itself. +// +// It fires only where the entity type *declares* monitoring. Types that +// declare `monitoring: none` (site, cluster, lan, mesh — topological groupings +// with nothing to probe) are working as intended and must never raise a +// signal; a permanent unresolvable warning against six healthy entities would +// discredit the whole thing. Types that declare nothing at all are a modelling +// gap, reported once per pass at debug level rather than as a fleet problem. +func coverageSweep(ctx context.Context, pool *db.Pool) { + // Resolution walks parent_type — inheriting types (lxc, proxmox-host, lan) + // carry NULL in their own monitoring_spec column, so reading it directly + // would flag every one of them. Reuse the Go resolver instead of + // duplicating the hierarchy walk in SQL. + tx, err := pool.Begin(ctx) + if err != nil { + slog.Error("scheduler: coverage sweep begin", "error", err) + return + } + tree, err := db.LoadTypeTree(ctx, tx) + if err != nil { + _ = tx.Rollback(ctx) + slog.Error("scheduler: coverage sweep load type tree", "error", err) + return + } + _ = tx.Rollback(ctx) // read-only + + rows, err := pool.Query(ctx, ` + SELECT e.id, e.slug, e.type, (cd.target_id IS NOT NULL) AS has_check + FROM entities e + LEFT JOIN ( + SELECT DISTINCT target_id FROM check_defs + WHERE enabled AND target_id IS NOT NULL + ) cd ON cd.target_id = e.id + WHERE e.state = 'active' AND e.type <> 'check'`) + if err != nil { + slog.Error("scheduler: coverage sweep query", "error", err) + return + } + + type entity struct { + id uuid.UUID + slug string + typ string + hasCheck bool + } + var all []entity + for rows.Next() { + var e entity + if err := rows.Scan(&e.id, &e.slug, &e.typ, &e.hasCheck); err != nil { + continue + } + all = append(all, e) + } + rows.Close() + if rows.Err() != nil { + slog.Error("scheduler: coverage sweep scan", "error", rows.Err()) + return + } + + var raised, resolved, undeclared int + for _, e := range all { + mon := tree.Monitoring(e.typ) + + switch { + case !mon.Declared: + undeclared++ + case mon.None(): + // Explicitly unmonitorable. Nothing to say. + case e.hasCheck: + if resolveCoverageSignal(ctx, pool, e.id) { + resolved++ + slog.Info("scheduler: entity is monitored again", "entity", e.slug) + } + default: + if raiseCoverageSignal(ctx, pool, e.id, e.slug, e.typ, mon.Kinds) { + raised++ + } + } + } + + if raised > 0 || resolved > 0 { + slog.Warn("scheduler: coverage sweep", + "unmonitored_raised", raised, "resolved", resolved, "scanned", len(all)) + } + if undeclared > 0 { + slog.Debug("scheduler: entity types declare no monitoring", "entities", undeclared) + } +} + +// raiseCoverageSignal raises (or refreshes) the unmonitored signal for one +// entity. Reports whether this was a new raise. +func raiseCoverageSignal(ctx context.Context, pool *db.Pool, entityID uuid.UUID, slug, typ string, want []string) bool { + // A signal is a dual entity: signals.entity_id is a PK referencing + // entities(id), so the row has to exist first. The scheduler's other + // signals borrow the check entity's id — there is no check here, which is + // the whole point, so this sweep owns a signal entity per target. + // + // The slug is stable per target, which makes the signal row stable too and + // lets a resolved signal be re-raised by primary key rather than colliding + // with it. + signalSlug := fmt.Sprintf("signal:%s:%s", UnmonitoredKind, slug) + + newID, err := uuid.NewV7() + if err != nil { + newID = uuid.New() + } + + var signalID uuid.UUID + // Upsert RETURNING id, never insert-and-assume: assuming is what made + // checkdefaults write foreign keys to rows it had not created. + if err := pool.QueryRow(ctx, + `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) + VALUES ($1, $2, 'signal', $2, 'active', '{}', 1, now(), now()) + ON CONFLICT (slug) DO UPDATE SET updated_at = now() + RETURNING id`, + newID, signalSlug).Scan(&signalID); err != nil { + slog.Error("scheduler: upsert signal entity", "entity", slug, "error", err) + return false + } + + evidence := fmt.Sprintf("type %s declares monitoring %v but the entity has no enabled check_def", typ, want) + + // Conflict on the primary key rather than on the (target, kind) partial + // index: that index only covers OPEN signals, so a previously resolved + // signal would not conflict there and would collide on the PK instead. + tag, err := pool.Exec(ctx, + `INSERT INTO signals (entity_id, kind, severity, target_entity_id, evidence, state) + VALUES ($1, $2, 'warning', $3, $4, 'raised') + ON CONFLICT (entity_id) DO UPDATE + SET state = CASE WHEN signals.state IN ('resolved','failed') THEN 'raised' ELSE signals.state END, + occurrence_count = signals.occurrence_count + 1, + evidence = EXCLUDED.evidence, + last_seen_at = now(), updated_at = now()`, + signalID, UnmonitoredKind, entityID, evidence) + if err != nil { + slog.Error("scheduler: raise unmonitored signal", "entity", slug, "error", err) + return false + } + + // RowsAffected is 1 for both insert and update, so ask the signal itself + // whether this was the first occurrence. + var occurrences int + if err := pool.QueryRow(ctx, + `SELECT occurrence_count FROM signals WHERE entity_id = $1`, signalID).Scan(&occurrences); err != nil { + return tag.RowsAffected() > 0 + } + if occurrences <= 1 { + slog.Warn("scheduler: entity is unmonitored", + "entity", slug, "type", typ, "declared", want) + emitSchedulerEvent(ctx, pool, "coverage.unmonitored", entityID, "warning", + map[string]any{"slug": slug, "type": typ, "declared": want}) + return true + } + return false +} + +// resolveCoverageSignal closes the unmonitored signal once the entity has a +// check. The scheduler's normal auto-resolve keys on the *check* entity id and +// only from state 'raised', so it can never clear one of these. +func resolveCoverageSignal(ctx context.Context, pool *db.Pool, entityID uuid.UUID) bool { + tag, err := pool.Exec(ctx, + `UPDATE signals SET state = 'resolved', updated_at = now() + WHERE target_entity_id = $1 AND kind = $2 + AND state NOT IN ('resolved', 'failed')`, + entityID, UnmonitoredKind) + if err != nil { + slog.Error("scheduler: resolve unmonitored signal", "error", err) + return false + } + return tag.RowsAffected() > 0 +} diff --git a/internal/scheduler/coverage_test.go b/internal/scheduler/coverage_test.go new file mode 100644 index 0000000..eb25bc9 --- /dev/null +++ b/internal/scheduler/coverage_test.go @@ -0,0 +1,309 @@ +package scheduler + +// Integration tests for coverageSweep against a real TimescaleDB. Guarded by +// OIKOS_TEST_DATABASE_URL — skipped when unset, same convention as +// internal/db/integration_test.go. Each run creates a throwaway database. + +import ( + "context" + "fmt" + "math/rand" + "os" + "strings" + "testing" + + "github.com/dtoro/oikos/internal/db" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +func newCoveragePool(t *testing.T) *db.Pool { + t.Helper() + baseURL := os.Getenv("OIKOS_TEST_DATABASE_URL") + if baseURL == "" { + t.Skip("OIKOS_TEST_DATABASE_URL not set — skipping integration test") + } + ctx := context.Background() + + admin, err := pgx.Connect(ctx, baseURL) + if err != nil { + t.Fatalf("connect admin: %v", err) + } + dbName := fmt.Sprintf("oikos_cov_%08x", rand.Int63()) + if _, err := admin.Exec(ctx, "CREATE DATABASE "+dbName); err != nil { + admin.Close(ctx) + t.Fatalf("create test db: %v", err) + } + admin.Close(ctx) + + at := strings.LastIndex(baseURL, "/") + testURL := baseURL[:at+1] + dbName + if q := strings.Index(baseURL[at:], "?"); q >= 0 { + testURL += baseURL[at+q:] + } + + pool, err := db.New(ctx, testURL) + if err != nil { + t.Fatalf("connect test db: %v", err) + } + if err := pool.Migrate(ctx); err != nil { + t.Fatalf("migrate: %v", err) + } + t.Cleanup(func() { + pool.Close() + if admin, err := pgx.Connect(ctx, baseURL); err == nil { + admin.Exec(ctx, "DROP DATABASE IF EXISTS "+dbName+" WITH (FORCE)") + admin.Close(ctx) + } + }) + return pool +} + +// fixture builds the four cases that matter, without the full seed: +// a declared+monitored type, a declared+unmonitored one, an explicitly +// unmonitorable one, and one that inherits its declaration from a parent. +func coverageFixture(t *testing.T, pool *db.Pool) map[string]uuid.UUID { + t.Helper() + ctx := context.Background() + + exec := func(sql string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("fixture %q: %v", sql, err) + } + } + + exec(`INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, monitoring_spec, status) + VALUES ('t-container', NULL, true, 'compute', 'infrastructure', '["resource"]', 'active') + ON CONFLICT (name) DO UPDATE SET monitoring_spec = EXCLUDED.monitoring_spec`) + exec(`INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, monitoring_spec, status) + VALUES ('t-lxc', 't-container', false, 'compute', 'infrastructure', NULL, 'active') + ON CONFLICT (name) DO NOTHING`) + exec(`INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, monitoring_spec, status) + VALUES ('t-service', NULL, false, 'software', 'infrastructure', '["http"]', 'active') + ON CONFLICT (name) DO NOTHING`) + exec(`INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, monitoring_spec, status) + VALUES ('t-site', NULL, false, 'physical', 'infrastructure', '[]', 'active') + ON CONFLICT (name) DO NOTHING`) + // `check` and `signal` normally arrive with the ontology seed, which this + // fixture skips; the sweep creates signal entities and needs both. + exec(`INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, monitoring_spec, status) + VALUES ('check', NULL, false, 'operations', 'cognition', '[]', 'active') + ON CONFLICT (name) DO NOTHING`) + exec(`INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, monitoring_spec, status) + VALUES ('signal', NULL, false, 'operations', 'cognition', '[]', 'active') + ON CONFLICT (name) DO NOTHING`) + + ids := map[string]uuid.UUID{} + for _, e := range []struct{ slug, typ string }{ + {"monitored-svc", "t-service"}, // declared + has a check + {"unmonitored-svc", "t-service"}, // declared + no check → signal + {"inheriting-lxc", "t-lxc"}, // inherits [resource], no check → signal + {"a-site", "t-site"}, // explicitly none → never a signal + } { + id := uuid.New() + exec(`INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) + VALUES ($1,$2,$3,$2,'active','{}',1,now(),now())`, id, e.slug, e.typ) + ids[e.slug] = id + } + + // Only monitored-svc gets a check. + checkID := uuid.New() + exec(`INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) + VALUES ($1,'check:t:monitored-svc:0','check','c-monitored-svc','active','{}',1,now(),now())`, checkID) + exec(`INSERT INTO check_defs (entity_id, target_id, kind, config, interval_s, timeout_s, enabled) + VALUES ($1,$2,'http','{}',60,30,true)`, checkID, ids["monitored-svc"]) + + return ids +} + +func openSignals(t *testing.T, pool *db.Pool, targetID uuid.UUID) (int, int) { + t.Helper() + var count, occurrences int + err := pool.QueryRow(context.Background(), + `SELECT count(*), COALESCE(max(occurrence_count),0) FROM signals + WHERE target_entity_id = $1 AND kind = $2 AND state NOT IN ('resolved','failed')`, + targetID, UnmonitoredKind).Scan(&count, &occurrences) + if err != nil { + t.Fatalf("count signals: %v", err) + } + return count, occurrences +} + +func TestCoverageSweepOnlyFlagsDeclaredButUnmonitored(t *testing.T) { + pool := newCoveragePool(t) + ids := coverageFixture(t, pool) + ctx := context.Background() + + coverageSweep(ctx, pool) + + for _, c := range []struct { + slug string + want int + why string + }{ + {"unmonitored-svc", 1, "declares http, has no check"}, + {"inheriting-lxc", 1, "inherits [resource] from its parent, has no check"}, + {"monitored-svc", 0, "has a check"}, + {"a-site", 0, "declares monitoring: none — flagging it would be a permanent false positive"}, + } { + got, _ := openSignals(t, pool, ids[c.slug]) + if got != c.want { + t.Errorf("%s (%s): %d open unmonitored signals, want %d", c.slug, c.why, got, c.want) + } + } +} + +func TestCoverageSweepDedupsRatherThanDuplicating(t *testing.T) { + pool := newCoveragePool(t) + ids := coverageFixture(t, pool) + ctx := context.Background() + + coverageSweep(ctx, pool) + coverageSweep(ctx, pool) + coverageSweep(ctx, pool) + + count, occurrences := openSignals(t, pool, ids["unmonitored-svc"]) + if count != 1 { + t.Errorf("three sweeps produced %d signals, want 1", count) + } + if occurrences != 3 { + t.Errorf("occurrence_count = %d after three sweeps, want 3", occurrences) + } +} + +func TestCoverageSweepResolvesWhenACheckAppears(t *testing.T) { + pool := newCoveragePool(t) + ids := coverageFixture(t, pool) + ctx := context.Background() + + coverageSweep(ctx, pool) + if count, _ := openSignals(t, pool, ids["unmonitored-svc"]); count != 1 { + t.Fatalf("expected an open signal before the check is added, got %d", count) + } + + // The entity acquires a check. The scheduler's normal auto-resolve keys on + // the check entity id and only from 'raised', so it could never clear this. + checkID := uuid.New() + if _, err := pool.Exec(ctx, + // entities has a unique (type, name), so this cannot reuse the + // fixture check's name. + `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) + VALUES ($1,'check:t:unmonitored-svc:0','check','c-unmonitored-svc','active','{}',1,now(),now())`, checkID); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, + `INSERT INTO check_defs (entity_id, target_id, kind, config, interval_s, timeout_s, enabled) + VALUES ($1,$2,'http','{}',60,30,true)`, checkID, ids["unmonitored-svc"]); err != nil { + t.Fatal(err) + } + + coverageSweep(ctx, pool) + + if count, _ := openSignals(t, pool, ids["unmonitored-svc"]); count != 0 { + t.Errorf("signal should have resolved once the entity had a check, %d still open", count) + } +} + +// Against the real ontology and inventory rather than a fixture: the sweep +// must stay silent for types that opted out and speak up for the genuine gaps. +// Asserted as properties, not exact counts, so it does not break every time +// the fleet changes. +func TestCoverageSweepAgainstTheRealSeed(t *testing.T) { + pool := newCoveragePool(t) + ctx := context.Background() + + for _, f := range []string{"ontology.yaml", "inventory.yaml", "policy.yaml"} { + content, err := os.ReadFile("../../seeds/" + f) + if err != nil { + t.Fatalf("read %s: %v", f, err) + } + err = pool.SeedIngest(ctx, f, content, + func(ctx context.Context, tx pgx.Tx, data map[string]any) error { + switch f { + case "ontology.yaml": + _, err := db.IngestOntologySeed(ctx, tx, data) + return err + case "inventory.yaml": + _, err := db.IngestInventorySeed(ctx, tx, data) + return err + default: + _, err := db.IngestPolicySeed(ctx, tx, data) + return err + } + }) + if err != nil { + t.Fatalf("ingest %s: %v", f, err) + } + } + + coverageSweep(ctx, pool) + + flagged := map[string]int{} + rows, err := pool.Query(ctx, ` + SELECT e.type, count(*) + FROM signals s JOIN entities e ON e.id = s.target_entity_id + WHERE s.kind = $1 AND s.state NOT IN ('resolved','failed') + GROUP BY e.type`, UnmonitoredKind) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var typ string + var n int + if err := rows.Scan(&typ, &n); err != nil { + t.Fatal(err) + } + flagged[typ] = n + } + rows.Close() + + t.Logf("unmonitored signals by entity type: %v", flagged) + + // Declared `monitoring: none` — flagging these would be a permanent, + // unresolvable false positive, which is the failure mode that would make + // the signal worthless. + for _, typ := range []string{"site", "lan", "mesh", "cluster"} { + if n := flagged[typ]; n != 0 { + t.Errorf("%s declares monitoring: none but %d were flagged unmonitored", typ, n) + } + } + + // Types that now get real checks must not be flagged either. + for _, typ := range []string{"service", "lxc", "ingress-route", "proxmox-host"} { + if n := flagged[typ]; n != 0 { + t.Errorf("%s should be covered by checkdefaults, but %d were flagged", typ, n) + } + } + + // Genuine gaps: no `dns` checker, no edge from a pool to its machine. + for _, typ := range []string{"dns-zone", "storage-pool"} { + if flagged[typ] == 0 { + t.Errorf("%s is a known gap and should have been flagged", typ) + } + } +} + +// A resolved signal must be re-raisable. The (target, kind) partial unique +// index only covers open signals, so a resolved row does not conflict there — +// it collides on the primary key instead, which is why the upsert targets the +// PK. +func TestCoverageSweepReRaisesAfterResolution(t *testing.T) { + pool := newCoveragePool(t) + ids := coverageFixture(t, pool) + ctx := context.Background() + + coverageSweep(ctx, pool) + if _, err := pool.Exec(ctx, + `UPDATE signals SET state='resolved' WHERE target_entity_id=$1 AND kind=$2`, + ids["unmonitored-svc"], UnmonitoredKind); err != nil { + t.Fatal(err) + } + + coverageSweep(ctx, pool) + + count, _ := openSignals(t, pool, ids["unmonitored-svc"]) + if count != 1 { + t.Errorf("a resolved signal should be re-raisable, got %d open", count) + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 96f02dc..90a6127 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -16,6 +16,7 @@ import ( "regexp" "runtime" "strconv" + "strings" "time" "github.com/dtoro/oikos/internal/config" @@ -73,7 +74,12 @@ func runCheckPass(ctx context.Context, pool *db.Pool) { return } if len(defs) == 0 { + // Still run housekeeping: a fleet with no enabled check_defs is + // precisely the case coverageSweep exists to report, and returning + // here would mean the one situation that most needs reporting is the + // one situation that stays silent. slog.Debug("scheduler: no enabled check_defs") + housekeeping(ctx, pool) return } @@ -248,6 +254,8 @@ func executeCheck(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) check return checkPing(ctx, cd) case "ssh-script": return checkSSHScript(ctx, cd) + case "backup-freshness": + return checkBackupFreshness(ctx, cd) default: return checkResult{health: "unknown"} } @@ -265,6 +273,7 @@ func housekeeping(ctx context.Context, pool *db.Pool) { } staleSweep(ctx, pool) + coverageSweep(ctx, pool) // Log housekeeping completion slog.Debug("scheduler: housekeeping done", "pruned_idempotency_before", cutoff.Format(time.RFC3339)) @@ -337,9 +346,14 @@ func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkRes cfg := struct { URL string `json:"url"` ExpectedStatus int `json:"expected_status"` - Insecure bool `json:"insecure"` + // MaxStatus accepts a range instead of one exact code. Most services + // sit behind Authentik and answer 302 or 401 — a working service, but + // an exact-match on 200 reports it degraded and raises a signal. + // Unset expected_status means "any response below MaxStatus is fine". + MaxStatus int `json:"max_status"` + Insecure bool `json:"insecure"` }{ - ExpectedStatus: 200, + MaxStatus: 500, } if len(cd.Config) > 0 { _ = json.Unmarshal(cd.Config, &cfg) @@ -379,10 +393,17 @@ func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkRes } defer resp.Body.Close() - if resp.StatusCode != cfg.ExpectedStatus { + if cfg.ExpectedStatus != 0 { + if resp.StatusCode != cfg.ExpectedStatus { + return checkResult{ + health: "degraded", signalKind: "http", + evidence: fmt.Sprintf("GET %s returned %d (expected %d)", cfg.URL, resp.StatusCode, cfg.ExpectedStatus), + } + } + } else if resp.StatusCode >= cfg.MaxStatus { return checkResult{ health: "degraded", signalKind: "http", - evidence: fmt.Sprintf("GET %s returned %d (expected %d)", cfg.URL, resp.StatusCode, cfg.ExpectedStatus), + evidence: fmt.Sprintf("GET %s returned %d (expected below %d)", cfg.URL, resp.StatusCode, cfg.MaxStatus), } } @@ -616,6 +637,11 @@ func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) che Port int `json:"port"` User string `json:"user"` Script string `json:"script"` + // Args is a single positional argument for the script. checkdefaults + // has always written it for process_check.sh, but nothing read it — + // so every process check ran argument-less and process_check.sh + // answered "no service name provided" with health unknown. + Args string `json:"args"` }{} if len(cd.Config) > 0 { _ = json.Unmarshal(cd.Config, &cfg) @@ -646,6 +672,11 @@ func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) che defer cancel() scriptPath := "/opt/oikos/checks/" + cfg.Script + if cfg.Args != "" { + // Single-quote the argument so an entity name can never break out of + // the remote command. The script name itself is allowlisted above. + scriptPath += " '" + strings.ReplaceAll(cfg.Args, "'", `'\''`) + "'" + } port := strconv.Itoa(cfg.Port) output, err := sshExec(ctx, cfg.Host, port, cfg.User, scriptPath, timeout) diff --git a/migrations/022_entity_type_monitoring.up.sql b/migrations/022_entity_type_monitoring.up.sql new file mode 100644 index 0000000..24e9318 --- /dev/null +++ b/migrations/022_entity_type_monitoring.up.sql @@ -0,0 +1,35 @@ +-- 022_entity_type_monitoring.up.sql +-- Declare, per entity type, what monitoring that type warrants. +-- +-- Motivation: only 3 of 89 active entities had an enabled check_def, because +-- checkdefaults.forEntityType hardcoded a Go `switch` over five entity types +-- and resolveHost looked for attribute shapes the seed data never used. The +-- failure was silent — every tx.Exec in that file discarded its error. +-- +-- Fixing coverage alone is not enough: coverage is NOT uniform. Some types +-- (site, cluster, lan, mesh) are topological groupings with nothing to probe; +-- their health is implied by their members. Without an explicit declaration, +-- the "unmonitored" signal added alongside this migration would fire +-- permanently and unresolvably against entities that are working as intended. +-- +-- So monitoring becomes a property of the TYPE, resolved through the existing +-- `parent_type` is-a hierarchy (declaring it on abstract `machine` covers +-- proxmox-host / standalone-server / workstation). +-- +-- Three states, deliberately distinguishable: +-- NULL — undeclared. An ontology gap; reported at info severity, +-- not as a fleet gap. This is why the column is nullable +-- rather than defaulting to '[]'. +-- '[]' — explicitly none. Excluded from coverage signalling. +-- '["http", ...]' — the check kinds this type warrants. +-- +-- The column holds check KINDS only. Deriving each check's config (host, +-- script, url, thresholds) stays in Go, in internal/checkdefaults — a config +-- template language in YAML is the natural follow-on, not this change. + +ALTER TABLE entity_types ADD COLUMN IF NOT EXISTS monitoring_spec JSONB; + +-- Kept on one line and free of semicolons: the migration runner splits on ';' +-- without tracking string literals, so both a newline and an inner semicolon +-- would truncate this statement mid-quote. +COMMENT ON COLUMN entity_types.monitoring_spec IS 'Check kinds this type warrants, resolved through parent_type. NULL means undeclared (an ontology gap), [] means explicitly unmonitorable, ["http","resource"] means declared kinds. Populated from seeds/ontology.yaml.'; diff --git a/migrations/023_executions_created_at_index.up.sql b/migrations/023_executions_created_at_index.up.sql new file mode 100644 index 0000000..7c821f6 --- /dev/null +++ b/migrations/023_executions_created_at_index.up.sql @@ -0,0 +1,18 @@ +-- 023_executions_created_at_index.up.sql +-- Support newest-first execution history. +-- +-- ListExecutions previously ordered by the target entity's slug, which is +-- neither useful for a history view nor unique enough to paginate on. It now +-- orders by (created_at DESC, entity_id DESC) -- the compound key the cursor +-- carries -- and executions had indexes only on target_entity_id and status. +-- +-- The same ordering backs /activity/recent, which was doing this unindexed. + +CREATE INDEX IF NOT EXISTS idx_executions_created_at + ON executions (created_at DESC, entity_id DESC); + +-- Per-entity history ("what has run against this host?") filters on the target +-- and then sorts, so give it a composite rather than making the planner sort +-- every row for a target with a long history. +CREATE INDEX IF NOT EXISTS idx_executions_target_created_at + ON executions (target_entity_id, created_at DESC); diff --git a/migrations/024_execution_logs.up.sql b/migrations/024_execution_logs.up.sql new file mode 100644 index 0000000..9489cda --- /dev/null +++ b/migrations/024_execution_logs.up.sql @@ -0,0 +1,43 @@ +-- 024_execution_logs.up.sql +-- Incremental command output for executions. +-- +-- Until now `executions.result` was a single JSONB blob written once, at the +-- terminal state: {"output": "...everything..."}. Two consequences: +-- +-- 1. Nothing could be seen while a command ran. A ten-minute apt upgrade +-- showed an empty row until it finished. +-- 2. On the sshExecTimeout path the output was discarded entirely — the +-- code returned "" — so the executions most worth inspecting (the ones +-- that hung) were the ones that left no trace at all. +-- +-- Chunks land here as they arrive. `executions.result` still gets the full +-- output at the end, so existing readers keep working unchanged and this +-- table is purely additive. + +CREATE TABLE IF NOT EXISTS execution_logs ( + execution_id UUID NOT NULL, + ts TIMESTAMPTZ NOT NULL DEFAULT now(), + -- Monotonic per execution. ts alone cannot order chunks: several arrive + -- within the same microsecond on a fast command. + seq INTEGER NOT NULL, + -- 'stdout' or 'stderr'. Both are also concatenated into the combined + -- output, matching what CombinedOutput used to return. + stream TEXT NOT NULL, + chunk TEXT NOT NULL, + PRIMARY KEY (execution_id, seq, ts) +); + +SELECT create_hypertable('execution_logs', 'ts', + chunk_time_interval => INTERVAL '7 days', if_not_exists => TRUE); + +-- The only query that matters: replay one execution's output in order. +CREATE INDEX IF NOT EXISTS idx_execution_logs_exec_seq + ON execution_logs (execution_id, seq); + +-- Matches the events table's 90 days. Command output is bulkier than events, +-- but keeping it exactly as long as the event stream that references it avoids +-- dangling 'execution.output' events pointing at rows that no longer exist. +DO $$ BEGIN + PERFORM add_retention_policy('execution_logs', INTERVAL '90 days'); +EXCEPTION WHEN OTHERS THEN NULL; +END $$; diff --git a/seeds/inventory.yaml b/seeds/inventory.yaml index cfc88e4..97e075d 100644 --- a/seeds/inventory.yaml +++ b/seeds/inventory.yaml @@ -201,7 +201,17 @@ entities: - {slug: "volume:media-local", type: volume, name: media-local, attributes: {path: /mnt/media_local}} - {slug: "backup:proton-drive", type: backup-target, name: proton-drive, - attributes: {provider: proton, encrypted: true}} + attributes: {provider: proton, encrypted: true, + path: /mnt/backup, + note: "rclone stages here before pushing to Proton; freshness is checked on lxc:rclone via the backs-up-to edge"}} + # The pre-deploy pg_dump written by scripts/deploy.sh on every push to main. + # It was the lab's only untracked backup: its failure path is `|| echo + # WARNING` inside the deploy script, so a broken dump was invisible until a + # rollback needed it. + - {slug: "backup:oikos-predeploy", type: backup-target, name: oikos-predeploy, + attributes: {provider: local, encrypted: false, + path: /opt/oikos/backups, + note: "pre-deploy pg_dump on the mac-mini; one per deployed SHA"}} # ─── Services ────────────────────────────────────────────────────── - {slug: "service:proxmox-ui", type: service, name: proxmox_ui, @@ -536,6 +546,10 @@ relationships: - {source: "lxc:romm", target: "pool:ludo-lvm", type: stores-on} - {source: "lxc:teddycloud", target: "pool:local-lvm-hubris", type: stores-on} - {source: "lxc:rclone", target: "backup:proton-drive", type: backs-up-to} + # The mac-mini writes the pre-deploy dumps, so it is also where the freshness + # check runs — checkdefaults resolves a backup-target's host by walking this + # edge backwards. + - {source: "ws:mac-mini", target: "backup:oikos-predeploy", type: backs-up-to} # ─── Governance ──────────────────────────────────────────────────── - {source: "person:dtoro", target: "agent:nomos", type: owns} diff --git a/seeds/ontology.yaml b/seeds/ontology.yaml index 4cdb4e0..12ab8ec 100644 --- a/seeds/ontology.yaml +++ b/seeds/ontology.yaml @@ -176,6 +176,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Physical location (home, VPS datacenter). + monitoring: none # topological — health is its members' attributes: {type: object, properties: {address: {type: string}}} ups: parent: entity @@ -183,6 +184,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Uninterruptible power supply. + monitoring: none # warranted, but no SNMP/NUT checker exists yet attributes: {type: object, properties: {vendor: {type: string}, va: {type: integer}}} sensor: parent: entity @@ -190,12 +192,14 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Environmental sensor. + monitoring: none # readings are metrics, not health peripheral: parent: entity domain: physical layer: infrastructure lifecycle: infrastructure description: Attached hardware (GPU, e-ink display, dongle). + monitoring: none # visible only through its host # ── Infrastructure / compute ── compute-entity: @@ -210,6 +214,8 @@ entity_types: domain: compute layer: infrastructure description: Physical machine. Always instantiated as a subtype. + monitoring: [ping, resource, updates] # inherited by proxmox-host / + # standalone-server / workstation / appliance attributes: type: object properties: @@ -266,6 +272,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Virtual machine. + monitoring: [ping] # no guest agent assumed; reachability only attributes: type: object properties: @@ -281,6 +288,7 @@ entity_types: domain: compute layer: infrastructure description: OS-level container (LXC or Docker). + monitoring: [resource] # inherited by lxc / docker-container attributes: type: object properties: {runtime: {type: string}} @@ -317,6 +325,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Hypervisor software running on a machine (PVE, KVM, OrbStack). + monitoring: none # the hosting machine's checks cover it attributes: type: object properties: {type: {type: string}, version: {type: string}} @@ -328,6 +337,8 @@ entity_types: domain: network layer: infrastructure description: A network things connect to. + monitoring: none # inherited by lan / mesh / vlan — a network's + # reachability is a property of its members lan: parent: network domain: network @@ -360,6 +371,7 @@ entity_types: layer: infrastructure description: Optional per-interface refinement (mac, ip). The seed uses coarse connects-via edges; interfaces can be backfilled later. + monitoring: none # covered by its machine's ping/resource checks attributes: {type: object, properties: {mac: {type: string}, ip: {type: string}}} dns-zone: parent: entity @@ -367,12 +379,15 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: DNS zone (e.g. split-horizon hubris.network). + monitoring: [dns] # NOTE: no `dns` checker exists yet — this is a + # real gap and coverageSweep will report it attributes: {type: object, properties: {zone: {type: string}, authority: {type: string}}} dns-record: parent: entity domain: network layer: infrastructure description: Individual DNS record. + monitoring: none # the zone is the unit of monitoring attributes: type: object properties: {name: {type: string}, record_type: {type: string}, value: {type: string}} @@ -382,6 +397,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Public hostname → upstream mapping (Caddy). + monitoring: [http] # end-to-end: exercises Caddy + DNS + TLS + upstream attributes: type: object properties: @@ -393,12 +409,14 @@ entity_types: domain: network layer: infrastructure description: TLS certificate. + monitoring: [cert-expiry] attributes: {type: object, properties: {issuer: {type: string}, expires: {type: string}}} firewall-rule: parent: entity domain: network layer: infrastructure description: Firewall / port-forward rule. + monitoring: none # declarative config, not a running thing # ── Infrastructure / storage ── storage-pool: @@ -407,6 +425,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Storage pool (LVM, ZFS, NFS). + monitoring: [capacity] attributes: type: object properties: @@ -419,6 +438,7 @@ entity_types: lifecycle: infrastructure description: Named volume / dataset within a pool. Mount details live as attributes on `mounts` edges. + monitoring: [capacity] attributes: {type: object, properties: {size_gb: {type: number}, path: {type: string}}} backup-target: parent: entity @@ -426,6 +446,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Where backups land (Proton Drive, PBS). + monitoring: [backup-freshness] # checker lands in Phase 4 attributes: {type: object, properties: {provider: {type: string}, encrypted: {type: boolean}}} dataset: parent: entity @@ -433,6 +454,7 @@ entity_types: layer: infrastructure description: Logical data collection worth tracking independently of its volume (e.g. paperless documents). + monitoring: none # its volume and owning service carry the checks # ── Infrastructure / software ── service: @@ -441,6 +463,8 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: A running service with consumers. + monitoring: [http, process] # http when it has a `url`, else a process check + # on the host resolved through its hosting edge attributes: type: object properties: @@ -457,12 +481,14 @@ entity_types: domain: software layer: infrastructure description: Deployed application/package a service runs. + monitoring: none # the service in front of it is the probe target attributes: {type: object, properties: {version: {type: string}}} config-repo: parent: entity domain: software layer: infrastructure description: Git repo holding tracked configuration. + monitoring: none # its Gitea service carries the checks attributes: type: object properties: {url: {type: string}, branch: {type: string}} @@ -471,6 +497,7 @@ entity_types: domain: software layer: infrastructure description: Automated deploy path (webhook → script). + monitoring: none # health is per-deploy, tracked as executions attributes: type: object properties: {trigger: {type: string}, target_path: {type: string}} @@ -479,12 +506,14 @@ entity_types: domain: software layer: infrastructure description: Managed package baseline for a host class. + monitoring: none # drift shows up via each host's updates check cluster: parent: entity domain: software layer: infrastructure lifecycle: infrastructure description: Proxmox cluster. + monitoring: none # topological — its member hosts carry the checks attributes: {type: object, properties: {quorum: {type: string}}} compose-stack: parent: entity @@ -492,6 +521,7 @@ entity_types: layer: infrastructure lifecycle: infrastructure description: Docker Compose stack (the Oikos OS itself is one). + monitoring: [process] attributes: {type: object, properties: {path: {type: string}}} # ── Infrastructure / external ── @@ -500,22 +530,26 @@ entity_types: domain: external layer: infrastructure description: Registered public domain. + monitoring: none # expiry is a calendar concern, not a probe attributes: {type: object, properties: {registrar: {type: string}, expires: {type: string}}} cloud-service: parent: entity domain: external layer: infrastructure description: External SaaS/cloud dependency. + monitoring: [http] # only when the entity carries a `url` isp-link: parent: entity domain: external layer: infrastructure description: Internet uplink. + monitoring: none # no probe target; reachability shows up fleet-wide vendor-dependency: parent: entity domain: external layer: infrastructure description: Vendor the lab depends on (registrar, IONOS, Proton). + monitoring: none # a commercial relationship, not a running thing # ── Governance / identity ── person: @@ -532,6 +566,8 @@ entity_types: layer: governance lifecycle: infrastructure # agents are deployed/retired like infrastructure description: Software agent actor (Nomos, the Oikos control loop). + monitoring: [http] # governance layer, but genuinely probeable — + # Nomos serves a gateway on :8092 attributes: type: object properties: diff --git a/seeds/policy.yaml b/seeds/policy.yaml index 82453a5..57de670 100644 --- a/seeds/policy.yaml +++ b/seeds/policy.yaml @@ -44,6 +44,23 @@ risk_classes: autonomy_allowed: false approval_rules: + # Signal kinds are looked up as `action` by internal/policy/classify.go, and + # an unmatched kind silently falls back to reversible_low/operator. Declare + # `unmonitored` so its routing is intentional: it reports a coverage gap and + # there is nothing to remediate automatically — closing it means an operator + # adding a check_def, which is its own deliberate change. + - {entity_type: entity, action: unmonitored, risk_class: read_only, autonomy_level: auto} + + # Backup freshness signals. Reporting-only for the same reason: the fix for a + # stale or missing backup is a deliberate human change (re-run the job, fix + # the mount, correct the path), never something to auto-remediate. Declared + # so the routing is intentional rather than the reversible_low/operator + # fallback an unmatched kind would otherwise get. + - {entity_type: backup-target, action: backup-stale, risk_class: read_only, autonomy_level: auto} + - {entity_type: backup-target, action: backup-missing, risk_class: read_only, autonomy_level: auto} + - {entity_type: backup-target, action: backup-misconfigured, risk_class: read_only, autonomy_level: auto} + - {entity_type: backup-target, action: backup-unreachable, risk_class: read_only, autonomy_level: auto} + # ── Generic rules on (possibly abstract) types ── - {entity_type: service, action: restart, risk_class: reversible_low, autonomy_level: auto} - {entity_type: service, action: cache-clear, risk_class: reversible_low, autonomy_level: auto} diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index b754962..09ae8eb 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -706,6 +706,25 @@ export async function fetchEntitySignals(entityId: string): Promise { return data.items ?? [] } +export interface ExecutionLogChunk { + seq: number + stream: string + chunk: string + ts: string +} + +// Streamed command output. Chunks land in execution_logs as the command runs, +// so this returns output for an execution that is still going — unlike +// `result.output`, which is only written once at the terminal state. +export async function fetchExecutionLogs( + executionId: string +): Promise<{ items: ExecutionLogChunk[]; combined: string }> { + const res = await fetchWithAuth(`${API}/executions/${executionId}/logs?limit=2000`) + if (!res.ok) return { items: [], combined: '' } + const data = await res.json() + return { items: data.items ?? [], combined: data.combined ?? '' } +} + export async function fetchEntityExecutions(entityId: string): Promise { const params = new URLSearchParams({ target: entityId, limit: '50' }) const res = await fetchWithAuth(`${API}/executions?${params}`) diff --git a/web/src/lib/components/EntityDetailContent.svelte b/web/src/lib/components/EntityDetailContent.svelte index ce9cbee..54580c2 100644 --- a/web/src/lib/components/EntityDetailContent.svelte +++ b/web/src/lib/components/EntityDetailContent.svelte @@ -10,6 +10,8 @@ fetchMetrics, fetchEntityEvents, fetchEntitySignals, + fetchEntityExecutions, + fetchExecutionLogs, fetchEntityTasks, fetchEntityKnowledge, fetchKnowledgeContent, @@ -24,6 +26,7 @@ type Relationship, type MetricSeries, type Signal, + type Execution, type EntityTask, type KnowledgeHit, type KnowledgeContent, @@ -32,7 +35,7 @@ type AuditEntry } from '$lib/api' import { relativeTime, truncateMiddle } from '$lib/utils' - import type { OikosEvent } from '$lib/stores/events' + import { liveEvents, subscribeEvents, type OikosEvent } from '$lib/stores/events' import DetailSection from '$lib/components/DetailSection.svelte' import { Badge } from '$lib/components/ui/badge' import { Button } from '$lib/components/ui/button' @@ -48,6 +51,15 @@ let metrics = $state([]) let events = $state([]) let signals = $state([]) + let executions = $state([]) + let expandedExecution = $state(null) + // Streamed output for the expanded execution, kept separate from + // result.output: result is only written at the terminal state, so a running + // command has nothing there and these chunks are the only thing to show. + let streamedOutput = $state('') + let streamEl = $state(null) + // Follow the tail unless the operator has scrolled up to read something. + let followTail = $state(true) let tasks = $state([]) let knowledge = $state([]) let ownContent = $state(null) @@ -76,11 +88,12 @@ loading = false return } - const [rel, m, ev, sig, tk, kh, oc, ch, aa, au] = await Promise.all([ + const [rel, m, ev, sig, ex, tk, kh, oc, ch, aa, au] = await Promise.all([ fetchEntityRelations(entity.id), fetchMetrics(entity.id), fetchEntityEvents(entity.id), fetchEntitySignals(entity.id), + fetchEntityExecutions(entity.id), fetchEntityTasks(entity), fetchEntityKnowledge(entity.id), KNOWLEDGE_TYPES.has(entity.type) ? fetchKnowledgeContent(entity.id) : Promise.resolve(null), @@ -92,6 +105,7 @@ metrics = m events = ev signals = sig + executions = ex tasks = tk knowledge = kh ownContent = oc @@ -141,8 +155,80 @@ } } + const RUNNING_STATUSES = new Set(['running', 'approved', 'executing', 'pending_approval']) + + function isRunning(execution: Execution): boolean { + return RUNNING_STATUSES.has(execution.status) + } + + async function loadExecutionLogs(executionId: string) { + const logs = await fetchExecutionLogs(executionId) + // Ignore a response that arrives after the operator collapsed the row or + // opened a different one. + if (expandedExecution !== executionId) return + streamedOutput = logs.combined + if (followTail) { + await tick() + if (streamEl) streamEl.scrollTop = streamEl.scrollHeight + } + } + + async function toggleExecution(execution: Execution) { + if (expandedExecution === execution.id) { + expandedExecution = null + streamedOutput = '' + return + } + expandedExecution = execution.id + streamedOutput = '' + followTail = true + await loadExecutionLogs(execution.id) + } + + function onStreamScroll() { + if (!streamEl) return + // Re-engage following once the operator scrolls back to the bottom. + followTail = streamEl.scrollHeight - streamEl.scrollTop - streamEl.clientHeight < 24 + } + + // Live tail. The backend throttles execution.output to one event per second + // per execution and the NOTIFY payload deliberately omits the data, so the + // event is only a "there is more" ping — the chunks are re-read here. + // + // Lifecycle events (execution.completed/failed) refresh the list instead: + // without that the row keeps its `running` badge and empty duration forever, + // which only became visible once running executions were shown at all. + $effect(() => { + const ev = $liveEvents[0] + if (!ev || !ev.type.startsWith('execution.')) return + + if (ev.type === 'execution.output') { + const target = expandedExecution + if (target && ev.entity_id === target) loadExecutionLogs(target) + return + } + + if (!entity) return + // Only refetch for an execution this panel is actually showing, so an + // unrelated command elsewhere in the fleet doesn't cause a request here. + if (executions.some((e) => e.id === ev.entity_id)) { + refreshExecutions() + } + }) + + async function refreshExecutions() { + if (!entity) return + const id = entity.id + const next = await fetchEntityExecutions(id) + // Guard against the panel having switched entity mid-flight. + if (entity?.id === id) executions = next + } + onMount(() => { load(slug) + // One shared, reference-counted SSE connection; this just registers + // interest so the tail receives events while the window is open. + return subscribeEvents() }) $effect(() => { @@ -171,6 +257,53 @@ } } + // The `run` tool encodes its action as `run:{"command":…,"purpose":…}`, so + // the raw string is unreadable. Show the command when there is one, the bare + // verb otherwise. Mirrors splitAction() in internal/httpapi/activity.go. + function executionSummary(execution: Execution): string { + const idx = execution.action.indexOf(':') + if (idx < 0) return execution.action + const verb = execution.action.slice(0, idx) + const rest = execution.action.slice(idx + 1) + try { + const params = JSON.parse(rest) + if (typeof params?.command === 'string') return params.command + if (typeof params?.purpose === 'string') return `${verb} — ${params.purpose}` + } catch { + // Not JSON — older actions use `verb:plain-params`. + return `${verb} ${rest}` + } + return verb + } + + // result is {"output": …} on success and {"output": …, "error": …} on + // failure. Until now nothing in the UI rendered either. + function executionOutput(execution: Execution): string { + const result = execution.result + if (!result) return '' + const parts: string[] = [] + if (typeof result.error === 'string' && result.error) parts.push(result.error) + if (typeof result.output === 'string' && result.output) parts.push(result.output) + return parts.join('\n\n').trim() + } + + function executionStatusVariant( + status: string + ): 'default' | 'secondary' | 'destructive' | 'outline' { + if (status === 'failed' || status === 'denied' || status === 'revoked') return 'destructive' + if (status === 'completed') return 'default' + if (status === 'running' || status === 'pending_approval') return 'secondary' + return 'outline' + } + + function formatDuration(ms: number | null): string { + if (ms == null) return '—' + if (ms < 1000) return `${ms}ms` + const s = Math.round(ms / 1000) + if (s < 60) return `${s}s` + return `${Math.floor(s / 60)}m ${s % 60}s` + } + function severityVariant(sev: string): 'default' | 'secondary' | 'destructive' { if (sev === 'critical') return 'destructive' if (sev === 'warning') return 'secondary' @@ -550,6 +683,64 @@ {/snippet} + {#snippet executionsContent()} +
+ {#each executions as execution (execution.id)} + {@const output = executionOutput(execution)} + {@const running = isRunning(execution)} + {@const expanded = expandedExecution === execution.id} +
+
+ +
+ {#if execution.duration_ms != null} + {formatDuration(execution.duration_ms)} + {:else if running && execution.started_at} + + {relativeTime(execution.started_at)} + {/if} + {execution.status} +
+
+
+ {relativeTime(execution.started_at ?? execution.created_at)} + · + {execution.risk_class} +
+ {#if expanded} + {@const shown = running ? streamedOutput : streamedOutput || output} + {#if shown} +
{shown}
+ {:else if running} +

Waiting for output…

+ {/if} + {#if running} +
+ + {followTail ? 'following output' : 'scrolled up — paused'} +
+ {/if} + {/if} +
+ {:else} +

None.

+ {/each} +
+ {/snippet} + {#snippet tasksContent()}
{#each tasks as { task, executionCount } (task.id)} @@ -674,6 +865,12 @@ }, { key: 'metrics', title: 'Metrics', count: metrics.length, content: metricsContent }, { key: 'signals', title: 'Signals', count: signals.length, content: signalsContent }, + { + key: 'executions', + title: 'Executions', + count: executions.length, + content: executionsContent + }, { key: 'tasks', title: 'Tasks', count: tasks.length, content: tasksContent }, { key: 'knowledge', title: 'Knowledge', count: knowledge.length, content: knowledgeContent }, { key: 'events', title: 'Recent events', count: events.length, content: eventsContent }, diff --git a/web/src/lib/components/UnifiedTimeline.svelte b/web/src/lib/components/UnifiedTimeline.svelte index 2dd7cd5..09548a2 100644 --- a/web/src/lib/components/UnifiedTimeline.svelte +++ b/web/src/lib/components/UnifiedTimeline.svelte @@ -53,6 +53,18 @@ stepToggles.set(step.id, !stepOpen(step)) stepToggles = new Map(stepToggles) } + // Pin each streaming output pane to its tail as chunks arrive. Keyed by + // tool id because several run entries can be on screen, though only the + // newest one is ever actually streaming. + let liveOutputEls = $state>({}) + $effect(() => { + for (const e of entries) { + if (!e.liveOutput) continue + const el = liveOutputEls[e.id] + if (el) el.scrollTop = el.scrollHeight + } + }) + function toggleTool(id: string) { if (expandedTools.has(id)) expandedTools.delete(id) else expandedTools.add(id) @@ -366,7 +378,7 @@ {#if expandedWithTools}
{#each item.tools as tool (tool.id)} - {@const tOpen = expandedTools.has(tool.id)} + {@const tOpen = expandedTools.has(tool.id) || !!tool.liveOutput}
(tool.args || tool.detail) && toggleTool(tool.id)} + onclick={() => + (tool.args || tool.detail || tool.liveOutput) && toggleTool(tool.id)} > {#if tool.status === 'running'} @@ -428,6 +442,13 @@ tool.args )} {/if} + {#if tool.liveOutput} + +
{tool.liveOutput}
+ {/if} {#if tool.detail}
 {
+    const store = mod.liveExecutionOutputFor('session-1')
+    const stop = store.subscribe(() => {})
+
+    liveEvents.set([event({ entity_id: 'exec-1' })])
+    await settle()
+    expect(get(store)).not.toBeNull()
+
+    liveEvents.set([event({ type: 'execution.completed', entity_id: 'exec-1' })])
+    await settle()
+    expect(get(store)).toBeNull()
+    stop()
+  })
+
+  it('does not clear on another session completing', async () => {
+    const store = mod.liveExecutionOutputFor('session-1')
+    const stop = store.subscribe(() => {})
+
+    liveEvents.set([event({ entity_id: 'exec-1' })])
+    await settle()
+
+    liveEvents.set([
+      event({ type: 'execution.completed', entity_id: 'exec-5', correlation_id: 'session-2' })
+    ])
+    await settle()
+    expect(get(store)).not.toBeNull()
+    stop()
+  })
+})
diff --git a/web/src/lib/stores/execstream.ts b/web/src/lib/stores/execstream.ts
new file mode 100644
index 0000000..8c887d9
--- /dev/null
+++ b/web/src/lib/stores/execstream.ts
@@ -0,0 +1,102 @@
+// Live command output for a chat session's currently-running execution.
+//
+// The chat renders a `run` tool call as "running" from the moment the tool_use
+// arrives until its tool_result comes back — and for an auto-run that gap IS
+// the command's runtime. Until now that window showed only the arguments; the
+// output appeared all at once at the end.
+//
+// Correlation works because MCP-initiated executions now carry the chat
+// session id as their correlation_id, and every execution.output event carries
+// that through. So an event can be matched to the session on screen without a
+// lookup. Nomos runs tools sequentially, so at most one execution is in flight
+// per session — no ambiguity about which entry the output belongs to.
+//
+// Only auto-runs stream. A gated run returns "execution queued" immediately,
+// so its tool entry is already `done` and the command executes minutes later
+// after approval — the entity detail window is where that one is watched.
+
+import { readable, type Readable } from 'svelte/store'
+import { liveEvents, subscribeEvents } from './events'
+import { fetchExecutionLogs } from '$lib/api'
+
+export interface LiveExecutionOutput {
+  executionId: string
+  output: string
+}
+
+const cache = new Map>()
+
+/**
+ * Live output for whichever execution this session is currently running.
+ * Resets when a different execution starts, so output from a previous command
+ * never bleeds into the next one's entry.
+ */
+export function liveExecutionOutputFor(sessionId: string): Readable {
+  const existing = cache.get(sessionId)
+  if (existing) return existing
+
+  const store = readable(null, (set) => {
+    let currentId: string | null = null
+    let inFlight = false
+    // Coalesce: a refetch already running means the next event's data will be
+    // covered by a single follow-up, rather than queueing a request per event.
+    let queued = false
+
+    async function refresh(executionId: string) {
+      if (inFlight) {
+        queued = true
+        return
+      }
+      inFlight = true
+      try {
+        const logs = await fetchExecutionLogs(executionId)
+        // Drop a response for an execution that is no longer current.
+        if (currentId === executionId) set({ executionId, output: logs.combined })
+      } finally {
+        inFlight = false
+        if (queued) {
+          queued = false
+          if (currentId) refresh(currentId)
+        }
+      }
+    }
+
+    const unsubscribeSSE = subscribeEvents()
+    const unsubscribeEvents = liveEvents.subscribe((events) => {
+      const ev = events[0]
+      if (!ev) return
+
+      // Lifecycle end: clear so the finished command's output stops being
+      // shown against a new "running" entry.
+      if (
+        ev.correlation_id === sessionId &&
+        (ev.type === 'execution.completed' ||
+          ev.type === 'execution.failed' ||
+          ev.type === 'execution.cancelled')
+      ) {
+        currentId = null
+        set(null)
+        return
+      }
+
+      if (ev.type !== 'execution.output') return
+      if (ev.correlation_id !== sessionId) return
+      if (!ev.entity_id) return
+
+      if (ev.entity_id !== currentId) {
+        currentId = ev.entity_id
+        set({ executionId: currentId, output: '' })
+      }
+      refresh(ev.entity_id)
+    })
+
+    return () => {
+      unsubscribeEvents()
+      unsubscribeSSE()
+      cache.delete(sessionId)
+    }
+  })
+
+  cache.set(sessionId, store)
+  return store
+}
diff --git a/web/src/pages/Ops.svelte b/web/src/pages/Ops.svelte
index efb537b..e5276c3 100644
--- a/web/src/pages/Ops.svelte
+++ b/web/src/pages/Ops.svelte
@@ -45,7 +45,12 @@
     const ev = $liveEvents[0]
     if (!ev) return
     if (ev.type.startsWith('approval.')) loadApprovals()
-    if (ev.type.startsWith('execution.')) loadActivity()
+    // execution.output is a "more command output arrived" ping for one
+    // execution, not a lifecycle change — it fires up to once a second per
+    // running command and changes nothing this table shows. Refetching the
+    // whole activity list on it would turn a chatty apt upgrade into a
+    // refetch storm.
+    if (ev.type.startsWith('execution.') && ev.type !== 'execution.output') loadActivity()
   })
 
   async function decide(id: string, decision: 'approve' | 'deny') {
diff --git a/web/vite.config.ts b/web/vite.config.ts
index c16561b..85edc57 100644
--- a/web/vite.config.ts
+++ b/web/vite.config.ts
@@ -32,6 +32,16 @@ function authProxy(target: string, rewrite?: (path: string) => string): ProxyOpt
   }
 }
 
+// Where `npm run dev` proxies to. The SPA is hardwired to same-origin in dev
+// (see the __OIKOS_DEV_TOKEN__ define below), so these targets — not
+// localStorage — decide which backend a dev session actually talks to. They
+// default to the local prod stack, which is what you want day to day; override
+// them to point a dev SPA at a scratch API without touching this file:
+//
+//   OIKOS_API_PROXY=http://127.0.0.1:8199 npm run dev
+const apiTarget = process.env.OIKOS_API_PROXY ?? 'http://localhost:8090'
+const nomosTarget = process.env.OIKOS_NOMOS_PROXY ?? 'http://localhost:8092'
+
 export default defineConfig({
   plugins: [tailwindcss(), svelte()],
   base: '/',
@@ -71,11 +81,11 @@ export default defineConfig({
   },
   server: {
     proxy: {
-      '/api': authProxy('http://localhost:8090'),
+      '/api': authProxy(apiTarget),
       // Production Caddy strips /agent before forwarding to nomos
       // (compose/caddy/Caddyfile.oikos handle_path /agent/*); match that
       // here so dev and prod agree on nomos's actual route paths.
-      '/agent': authProxy('http://localhost:8092', (path) => path.replace(/^\/agent/, ''))
+      '/agent': authProxy(nomosTarget, (path) => path.replace(/^\/agent/, ''))
     }
   },
   test: {