Files
oikos/internal/policy/classify.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

161 lines
4.7 KiB
Go

// Package policy implements Oikos classification and policy evaluation.
// Determines risk class, autonomy route, and approval requirements.
package policy
import (
"context"
"encoding/json"
"fmt"
"github.com/dtoro/oikos/internal/domain"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
// ClassificationResult holds the outcome of classifying a signal.
type ClassificationResult struct {
RiskClass string
Route string // 'auto-act', 'escalate', 'hold'
RecommendedAction json.RawMessage
AutonomyCheck string
BlastRadius []uuid.UUID
CorrelationID string
Reasoning json.RawMessage
}
// Classify evaluates a signal against policy rules to determine the action route.
// ctx must have a DB connection pool accessible via a helper interface.
type Classifier struct {
DB interface {
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
Exec(ctx context.Context, sql string, args ...any) (int64, error)
}
}
// NewClassifier creates a classifier with a DB query interface.
func NewClassifier(dbc interface {
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
Exec(ctx context.Context, sql string, args ...any) (int64, error)
}) *Classifier {
return &Classifier{DB: dbc}
}
// ClassifySignal evaluates a signal and returns the classification result.
func (c *Classifier) ClassifySignal(ctx context.Context, signalEntityID, targetEntityID uuid.UUID,
kind, severity, correlationID string) (*ClassificationResult, error) {
// Determine target entity type
var entityType string
err := c.DB.QueryRow(ctx,
"SELECT type FROM entities WHERE id = $1", targetEntityID).Scan(&entityType)
if err != nil {
return nil, fmt.Errorf("%w: target entity %s", domain.ErrNotFound, targetEntityID)
}
// Look up risk class for this entity type and action
var riskClass string
var approvalRequired string
err = c.DB.QueryRow(ctx, `
SELECT rc.name, rc.approval_required
FROM risk_classes rc
WHERE rc.name = (
SELECT COALESCE(ar.risk_class, 'reversible_low')
FROM approval_rules ar
WHERE ar.entity_type = $1 AND ar.action = $2
LIMIT 1
)`, entityType, kind).Scan(&riskClass, &approvalRequired)
if err != nil {
// Default to escalate
riskClass = "reversible_low"
approvalRequired = "operator"
}
// Check global autonomy setting
var globalAutoAct string
err = c.DB.QueryRow(ctx,
"SELECT value FROM autonomy_settings WHERE key = 'global.auto_act'").Scan(&globalAutoAct)
if err != nil {
globalAutoAct = "on" // default to on
}
// Check per-entity kill-switch
var slug string
c.DB.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", targetEntityID).Scan(&slug)
var entityAutoAct string
if slug != "" {
c.DB.QueryRow(ctx,
"SELECT value FROM autonomy_settings WHERE key = $1",
"never_auto_act."+slug).Scan(&entityAutoAct)
}
// Determine route
route := "escalate"
autonomyCheck := ""
if globalAutoAct == "off" || globalAutoAct == "false" {
route = "escalate"
autonomyCheck = "blocked: global auto_act disabled"
} else if entityAutoAct == "true" {
route = "escalate"
autonomyCheck = "blocked: per-entity kill-switch"
} else if approvalRequired == "none" {
route = "auto-act"
autonomyCheck = "allowed"
} else {
autonomyCheck = "requires approval: " + approvalRequired
}
// Compute blast radius
blastRadius := computeBlastRadius(ctx, c.DB, targetEntityID)
reasoning := map[string]any{
"entity_type": entityType,
"risk_class": riskClass,
"approval_rule": approvalRequired,
"global_auto_act": globalAutoAct,
"entity_slug": slug,
}
reasoningJSON, _ := json.Marshal(reasoning)
recommended, _ := json.Marshal(map[string]any{
"action": kind,
"reason": fmt.Sprintf("signal %s on %s", severity, entityType),
})
return &ClassificationResult{
RiskClass: riskClass,
Route: route,
RecommendedAction: recommended,
AutonomyCheck: autonomyCheck,
BlastRadius: blastRadius,
CorrelationID: correlationID,
Reasoning: reasoningJSON,
}, nil
}
// computeBlastRadius traverses relationships to find affected entities.
func computeBlastRadius(ctx context.Context, dbc interface {
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
}, entityID uuid.UUID) []uuid.UUID {
rows, err := dbc.Query(ctx, `
SELECT entity_id FROM blast_radius($1, 3)`, entityID)
if err != nil {
return nil
}
defer rows.Close()
var ids []uuid.UUID
for rows.Next() {
var id uuid.UUID
if err := rows.Scan(&id); err == nil {
ids = append(ids, id)
}
}
return ids
}
// Ensure domain is used
var _ = domain.ErrAutonomyBlocked