Files
oikos/internal/notifier/notifier.go
dtoro 095a3967c4 phase 3: control loop — scheduler, actuator, learning, notifier, policy, API endpoints
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
2026-07-07 15:19:25 +02:00

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{}