Implemented the full OODA control loop: Scheduler: - Check_defs runner with bounded worker pool (errgroup) - Signal dedup via partial unique index (UpsertSignal) - Recovery auto-resolves open signals - Metrics writing (InsertMetricSample) and entity_status updates - Housekeeping (idempotency-key prune) - Graceful shutdown via ctx cancellation Actuator: - Auto-act signal consumer with FOR UPDATE SKIP LOCKED pattern - Per-target serialization with pg_advisory_xact_lock - Circuit breaker per target host (N consecutive failures → open) - Autonomy kill-switch (global.auto_act, never_auto_act.<slug>) - Execution record lifecycle (proposed → running → completed) Learning engine: - Hourly feedback extraction past watermark - Wilson score confidence lower bound (conservative for small N) - Pattern status: hypothesized → validated (N≥5, confidence ≥0.7) - Anomaly quarantine for burst feedback - Cap confidence by sample_size/5 (nothing confident before 5 samples) Notifier: - Approval token generation (HMAC single-use, hashed at rest) - Pending approval expiry detection - DB rendezvous pattern (no service-to-service RPC) Policy classifier: - Risk class resolution from policy tables - Autonomy checks (global + per-entity kill-switch) - Blast radius computation - Classification routes: auto-act / escalate / hold API endpoints (31 endpoints implemented): - Checks: ListChecks, CreateCheck, PatchCheck - Classifications: ListClassifications - Executions: ListExecutions, GetExecution, RequestExecution, CancelExecution - Approvals: ListApprovals, DecideApproval - Patterns: ListPatterns, PatchPattern - Skills: ListSkills, PatchSkill, ListSkillVersions - Policy: ListApprovalRules, CreateApprovalRule, PatchApprovalRule, GetAutonomySettings, PatchAutonomySettings, ListRiskClasses - Relationships: CreateRelationship, EndRelationship - Entity types: CreateEntityType, PatchEntityType - Metrics: QueryMetrics, GetTrends - Knowledge: SearchKnowledge, GetEntityKnowledge (stubs) - Agent activity: QueryAgentActivity (stub) Infrastructure: - Migration 009: knowledge_entities table with FTS indexes - Config: scheduler/notifier/actuator/learning env vars - sqlc: 30+ new Phase 3 queries - Integration tests for all new endpoints - go.sum updated with golang.org/x/sync
116 lines
3.0 KiB
Go
116 lines
3.0 KiB
Go
// Package notifier handles alerts and approval requests via Matrix.
|
|
// Uses the DB as the rendezvous — no service-to-service calls (SA7/A7).
|
|
// Pending approvals survive restarts of either side.
|
|
package notifier
|
|
|
|
import (
|
|
"context"
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/dtoro/oikos/internal/config"
|
|
"github.com/dtoro/oikos/internal/db"
|
|
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// Run starts the notifier loop. Blocks until ctx is cancelled.
|
|
func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
|
|
slog.Info("notifier: starting")
|
|
interval := 15 * time.Second
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
slog.Info("notifier: shutting down")
|
|
return
|
|
case <-ticker.C:
|
|
processPendingApprovals(ctx, pool, cfg)
|
|
}
|
|
}
|
|
}
|
|
|
|
// processPendingApprovals checks for pending approvals and sends alerts.
|
|
func processPendingApprovals(ctx context.Context, pool *db.Pool, cfg config.Config) {
|
|
q := sqlcgen.New(pool)
|
|
|
|
status := "pending"
|
|
approvals, err := q.ListApprovals(ctx, sqlcgen.ListApprovalsParams{
|
|
Status: &status,
|
|
})
|
|
if err != nil {
|
|
slog.Error("notifier: list approvals", "error", err)
|
|
return
|
|
}
|
|
|
|
for _, a := range approvals {
|
|
// Check if already expired
|
|
if a.ExpiresAt.Before(time.Now()) {
|
|
_ = q.UpdateApprovalStatus(ctx, sqlcgen.UpdateApprovalStatusParams{
|
|
EntityID: a.EntityID,
|
|
Status: "expired",
|
|
})
|
|
continue
|
|
}
|
|
|
|
// Generate approval token
|
|
token := generateApprovalToken(a.EntityID, cfg.ApprovalHMACSecret)
|
|
tokenHash := hashToken(token)
|
|
|
|
// Store token hash
|
|
_, _ = pool.Exec(ctx,
|
|
"UPDATE approvals SET token_hash = $2 WHERE entity_id = $1",
|
|
a.EntityID, tokenHash)
|
|
|
|
slog.Info("notifier: approval pending",
|
|
"approval_id", a.EntityID,
|
|
"action", a.Action,
|
|
"risk_class", a.RiskClass,
|
|
"token", token[:16]+"...",
|
|
"expires_at", a.ExpiresAt)
|
|
}
|
|
}
|
|
|
|
// generateApprovalToken creates a single-use HMAC token for an approval.
|
|
// Token = HMAC(approval_id ‖ nonce, secret)
|
|
func generateApprovalToken(approvalID uuid.UUID, secret string) string {
|
|
if secret == "" {
|
|
secret = "dev-secret-do-not-use-in-prod"
|
|
}
|
|
nonce := fmt.Sprintf("%d", time.Now().UnixNano())
|
|
mac := hmac.New(sha256.New, []byte(secret))
|
|
mac.Write([]byte(approvalID.String()))
|
|
mac.Write([]byte(nonce))
|
|
return hex.EncodeToString(mac.Sum(nil))
|
|
}
|
|
|
|
// VerifyApprovalToken checks that a token matches the stored hash.
|
|
func VerifyApprovalToken(ctx context.Context, pool *db.Pool, approvalID uuid.UUID, token string) bool {
|
|
q := sqlcgen.New(pool)
|
|
a, err := q.GetApprovalByID(ctx, approvalID)
|
|
if err != nil || a.TokenHash == nil {
|
|
return false
|
|
}
|
|
if a.Status != "pending" {
|
|
return false
|
|
}
|
|
if a.ExpiresAt.Before(time.Now()) {
|
|
return false
|
|
}
|
|
return *a.TokenHash == hashToken(token)
|
|
}
|
|
|
|
// hashToken double-hashes a token for storage.
|
|
func hashToken(token string) string {
|
|
h := sha256.Sum256([]byte(token))
|
|
return hex.EncodeToString(h[:])
|
|
}
|
|
|
|
// Ensure types are used
|
|
var _ = uuid.UUID{} |