feat: Phase 4 governance/execution slice — PolicyService + ExecutionService

classifyAndGate's decision pipeline moves to core: PolicyService runs the full gate order (classify + transport escalation, plan-first, syntax, host-only/host-lxc, VM QGA preflight, dedup, approval-flood, window routing) over ports.GovernanceStore; ExecutionService records and dispatches (auto-run via ssh.CommandExecutor + TargetResolver, queue via ExecutionRecorder) with one converged path for run/docker_exec. Gating matrix test added (risk x window x declared risk -> outcome); pair coverage 95.6%. Bug fix surfaced by the matrix: the flag-space syntax regex was inverted — it refused valid 'tail -n 3' and missed the actual 'head - n' typo. Fixed to match dash-space-value only. Remaining Phase 4 items tracked in the plan: ApprovalService.Decide convergence, execlog fold, execworker poller. VERSION 0.35.0.
2026-08-16 09:48:26 +02:00
parent 60c0432d8b
commit 7c9f4ec79f
19 changed files with 1748 additions and 725 deletions
--- a/internal/adapters/postgres/governance.go
+++ b/internal/adapters/postgres/governance.go
@@ -0,0 +1,323 @@
+package db
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/dtoro/oikos/internal/adapters/postgres/sqlcgen"
+	"github.com/dtoro/oikos/internal/core/domain"
+	"github.com/dtoro/oikos/internal/core/ports"
+	"github.com/dtoro/oikos/internal/observability"
+	"github.com/google/uuid"
+)
+
+// GovernanceRepo implements ports.GovernanceStore over the postgres
+// pool. The SQL is lifted from the MCP classifyAndGate path (Phase 4
+// convergence); fail-open/fail-closed semantics per gate are preserved
+// and documented on the port.
+type GovernanceRepo struct {
+	pool *Pool
+}
+
+var _ ports.GovernanceStore = (*GovernanceRepo)(nil)
+
+// NewGovernanceRepo builds the governance read model.
+func NewGovernanceRepo(pool *Pool) *GovernanceRepo { return &GovernanceRepo{pool: pool} }
+
+// SessionHasPlan fails open (true) on query error — a transient DB issue
+// must not block an otherwise-valid run.
+func (g *GovernanceRepo) SessionHasPlan(ctx context.Context, sessionID string) bool {
+	if sessionID == "" {
+		return true // no session → no gate (direct MCP call from a script)
+	}
+	var count int
+	if err := g.pool.QueryRow(ctx,
+		`SELECT COUNT(*) FROM session_plan_steps
+		 WHERE session_id = $1 AND status <> 'replaced'`,
+		sessionID).Scan(&count); err != nil {
+		return true
+	}
+	return count > 0
+}
+
+// AssentWindowActive checks the session-scoped assent window key set by
+// chat assent and the approval-decide path.
+func (g *GovernanceRepo) AssentWindowActive(ctx context.Context, agentID domain.UUID, sessionID string) bool {
+	if agentID == "" || sessionID == "" {
+		return false // fail closed
+	}
+	return g.windowActive(ctx, "assent_window.agent:"+string(agentID)+".session:"+sessionID)
+}
+
+// DestructiveWindowActive checks the target+session-scoped destructive
+// window key. Key format must match cmd/nomos/store.go's
+// openDestructiveWindow — both processes read/write the same rows.
+func (g *GovernanceRepo) DestructiveWindowActive(ctx context.Context, agentID domain.UUID, targetSlug, sessionID string) bool {
+	if agentID == "" || targetSlug == "" || sessionID == "" {
+		return false // fail closed
+	}
+	return g.windowActive(ctx,
+		"destructive_window.agent:"+string(agentID)+".target:"+targetSlug+".session:"+sessionID)
+}
+
+func (g *GovernanceRepo) windowActive(ctx context.Context, key string) bool {
+	var expiresStr string
+	err := g.pool.QueryRow(ctx,
+		"SELECT value FROM autonomy_settings WHERE key = $1", key).Scan(&expiresStr)
+	if err != nil {
+		return false
+	}
+	expires, err := time.Parse(time.RFC3339, expiresStr)
+	if err != nil {
+		return false
+	}
+	return time.Now().UTC().Before(expires)
+}
+
+// PendingApprovalCount returns the session's executions at pending_approval.
+func (g *GovernanceRepo) PendingApprovalCount(ctx context.Context, sessionID string) int {
+	var n int
+	if err := g.pool.QueryRow(ctx, `
+		SELECT COUNT(*) FROM nomos_plan_executions pe
+		JOIN executions ex ON ex.entity_id = pe.execution_id
+		WHERE pe.session_id = $1 AND ex.status = 'pending_approval'`,
+		sessionID).Scan(&n); err != nil {
+		return 0 // fail open for the flood gate: don't block on a flake
+	}
+	return n
+}
+
+// PendingDuplicate returns the newest identical pending execution on the
+// target, if any.
+func (g *GovernanceRepo) PendingDuplicate(ctx context.Context, targetID domain.UUID, action string) (domain.UUID, bool) {
+	var existingID string
+	err := g.pool.QueryRow(ctx, `
+		SELECT e.id::text FROM entities e
+		JOIN executions ex ON ex.entity_id = e.id
+		WHERE e.type = 'execution' AND ex.target_entity_id = $1
+		  AND ex.action = $2 AND ex.status = 'pending_approval'
+		ORDER BY e.created_at DESC LIMIT 1`,
+		mustUUID(targetID), action).Scan(&existingID)
+	if err != nil || existingID == "" {
+		return "", false
+	}
+	return domain.UUID(existingID), true
+}
+
+// EntityAttributes loads one entity's attributes map.
+func (g *GovernanceRepo) EntityAttributes(ctx context.Context, targetID domain.UUID) (map[string]any, error) {
+	var rawAttrs []byte
+	if err := g.pool.QueryRow(ctx,
+		`SELECT attributes FROM entities WHERE id = $1`, mustUUID(targetID)).Scan(&rawAttrs); err != nil {
+		return nil, err
+	}
+	var attrs map[string]any
+	if err := json.Unmarshal(rawAttrs, &attrs); err != nil {
+		return nil, err
+	}
+	return attrs, nil
+}
+
+// ResolveProxmoxHostSlug resolves the Proxmox host owning a guest slug
+// (refusal-hint input for host-only commands). Removed — importing
+// internal/remote from the postgres adapter forms an import cycle; the
+// hint is wired at composition time via PolicyService.HostHint instead.
+
+// ExecRunRepo implements ports.ExecutionRecorder over the postgres
+// pool: the write half of the run-submission lifecycle, lifted from the
+// MCP classifyAndGate / autoRun / createApproval /
+// markSessionAwaitingApproval paths (Phase 4 convergence).
+type ExecRunRepo struct {
+	pool *Pool
+}
+
+var _ ports.ExecutionRecorder = (*ExecRunRepo)(nil)
+
+// NewExecRunRepo builds the recorder.
+func NewExecRunRepo(pool *Pool) *ExecRunRepo { return &ExecRunRepo{pool: pool} }
+
+// CreateRun writes the execution entity + row, graph edges, session
+// link, classification row, and audit entry.
+func (r *ExecRunRepo) CreateRun(ctx context.Context, in ports.CreateRunInput) error {
+	id := mustUUID(in.ExecutionID)
+	execName := "run on " + in.TargetSlug + " (" + id.String() + ")"
+	execSlug := "exec:" + in.TargetSlug + ":" + id.String()
+	if _, err := r.pool.Exec(ctx,
+		`INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
+		id, execSlug, execName); err != nil {
+		return err
+	}
+	if _, err := r.pool.Exec(ctx,
+		`INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id)
+		 VALUES ($1, $2, $3, $4, 'running', $5, $6) ON CONFLICT DO NOTHING`,
+		id, mustUUID(in.TargetID), in.Action, in.RiskClass, in.CorrelationID, agentUUID(in.AgentID)); err != nil {
+		return err
+	}
+	if _, err := r.pool.Exec(ctx, `
+		INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
+		SELECT $1, $2, 'targets', '{"by":"nomos"}'::jsonb, now()
+		WHERE NOT EXISTS (
+			SELECT 1 FROM relationships
+			WHERE source_id = $1 AND target_id = $2 AND type = 'targets' AND valid_to IS NULL)`,
+		id, mustUUID(in.TargetID)); err != nil {
+		return err
+	}
+
+	auditSessionID := sessionUUID(in.SessionID)
+	if in.SessionID != "" {
+		if _, err := r.pool.Exec(ctx, `
+			INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
+			SELECT t.id, $1, 'involves', '{"by":"nomos"}'::jsonb, now()
+			FROM entities t WHERE t.slug = $2
+			AND NOT EXISTS (
+				SELECT 1 FROM relationships
+				WHERE source_id = t.id AND target_id = $1 AND type = 'involves' AND valid_to IS NULL)`,
+			id, "task:"+in.SessionID); err != nil {
+			return err
+		}
+		// Link execution to session for auto-continuation.
+		if auditSessionID != nil {
+			if _, err := r.pool.Exec(ctx, `
+				INSERT INTO nomos_plan_executions (execution_id, session_id)
+				VALUES ($1, $2) ON CONFLICT (execution_id) DO NOTHING`, id, *auditSessionID); err != nil {
+				return err
+			}
+		}
+	}
+
+	// Auto-classify: write the classification decision.
+	classID, err := uuid.NewV7()
+	if err != nil {
+		return err
+	}
+	if _, err := r.pool.Exec(ctx,
+		`INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'classification', $3, '{}')`,
+		classID, "classification:"+classID.String(), "classification for "+execSlug); err != nil {
+		return err
+	}
+	if _, err := r.pool.Exec(ctx,
+		`INSERT INTO classifications (entity_id, action, risk_class, route, reasoning, correlation_id)
+		 VALUES ($1, $2, $3, $4, $5, $6)`,
+		classID, in.Action, in.RiskClass, in.Route, string(in.ClassReasoning), in.CorrelationID); err != nil {
+		return err
+	}
+	if _, err := r.pool.Exec(ctx,
+		`UPDATE executions SET classification_id = $2 WHERE entity_id = $1`, id, classID); err != nil {
+		return err
+	}
+	if _, err := r.pool.Exec(ctx, `
+		INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
+		SELECT $1, $2, 'precedes', '{"by":"nomos"}'::jsonb, now()
+		WHERE NOT EXISTS (
+			SELECT 1 FROM relationships
+			WHERE source_id = $1 AND target_id = $2 AND type = 'precedes' AND valid_to IS NULL)`,
+		classID, id); err != nil {
+		return err
+	}
+
+	// Audit the run request (every run, auto or queued).
+	q := sqlcgen.New(r.pool)
+	return observability.Audit(ctx, q, "agent", "nomos", "run",
+		&id, "POST", "/mcp", in.CorrelationID, auditSessionID, map[string]any{
+			"command": in.Command, "target": in.TargetSlug,
+			"risk_class": in.RiskClass, "purpose": in.Purpose,
+		})
+}
+
+// MarkRunning stamps status + started_at now.
+func (r *ExecRunRepo) MarkRunning(ctx context.Context, id domain.UUID) error {
+	_, err := r.pool.Exec(ctx,
+		`UPDATE executions SET status='running', started_at=$2 WHERE entity_id=$1`,
+		mustUUID(id), time.Now())
+	return err
+}
+
+// Finalize stamps the terminal status, result payload, duration, and
+// completion time.
+func (r *ExecRunRepo) Finalize(ctx context.Context, in ports.FinalizeExecutionInput) error {
+	_, err := r.pool.Exec(ctx,
+		`UPDATE executions SET status=$2, result=$3::jsonb, duration_ms=$4,
+		        started_at=$5, completed_at=now()
+		 WHERE entity_id=$1`,
+		mustUUID(in.ExecutionID), in.Status, string(in.Result),
+		int(time.Since(in.StartedAt).Milliseconds()), in.StartedAt)
+	return err
+}
+
+// QueueApproval creates the approval row, flips the execution to
+// pending_approval, emits approval.created, and marks the session
+// awaiting_input.
+func (r *ExecRunRepo) QueueApproval(ctx context.Context, in ports.QueueApprovalInput) error {
+	id := mustUUID(in.ExecutionID)
+	payload, _ := json.Marshal(map[string]any{"action": in.Action, "params": in.Params, "execution_id": in.ExecutionID})
+	// approvals.entity_id is PK + FK to entities(id). Reuse the
+	// execution's entity so the FK is satisfied — a fresh UUID here has no
+	// matching entities row and the INSERT silently fails, orphaning the
+	// execution (one execution maps to at most one approval, so the 1:1
+	// identity holds).
+	subject := mustUUID(in.TargetID)
+	if err := sqlcgen.New(r.pool).InsertApproval(ctx, sqlcgen.InsertApprovalParams{
+		EntityID:        id,
+		SubjectEntityID: &subject,
+		Action:          in.Action,
+		RiskClass:       in.RiskClass,
+		Kind:            "execution",
+		Payload:         payload,
+		TokenHash:       nil,
+		ExpiresAt:       time.Now().Add(time.Hour),
+	}); err != nil {
+		return err
+	}
+	if _, err := r.pool.Exec(ctx,
+		`UPDATE executions SET approval_id = $1, status='pending_approval', risk_class=$2 WHERE entity_id = $1`,
+		id, in.RiskClass); err != nil {
+		return err
+	}
+
+	// Operator-facing SSE event: a gated action now awaits a decision.
+	q := sqlcgen.New(r.pool)
+	_ = observability.Event(ctx, q, "approval.created", &id, "warning", "mcp", "",
+		map[string]any{"action": in.Action, "params": in.Params, "risk_class": in.RiskClass})
+
+	// Flip the session to awaiting_input so boards and the idle sweep see
+	// the task as blocked on the operator, not silently working. No-op for
+	// direct MCP calls with no nomos session.
+	if in.SessionID == "" || in.SessionID == "ephemeral" {
+		return nil
+	}
+	tag, err := r.pool.Exec(ctx, `
+		UPDATE agent_sessions SET status = 'awaiting_input', last_active_at = now()
+		WHERE id = $1 AND status IN ('active', 'planning', 'executing')`, in.SessionID)
+	if err != nil || tag.RowsAffected() == 0 {
+		return nil
+	}
+	var taskEntID *uuid.UUID
+	var e uuid.UUID
+	if qerr := r.pool.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, in.SessionID).Scan(&e); qerr == nil && e != uuid.Nil {
+		taskEntID = &e
+	}
+	return observability.Event(ctx, q, "task.status", taskEntID, "info", "nomos", in.SessionID,
+		map[string]any{"status": "awaiting_input", "reason": "execution_pending_approval"})
+}
+
+// agentUUID maps a domain agent id to pg uuid (nil-safe).
+func agentUUID(id domain.UUID) any {
+	if id == "" {
+		return nil
+	}
+	return mustUUID(id)
+}
+
+// sessionUUID parses a session id for audit linkage; nil when absent or
+// ephemeral.
+func sessionUUID(sessionID string) *uuid.UUID {
+	if sessionID == "" || sessionID == "ephemeral" {
+		return nil
+	}
+	sid, err := uuid.Parse(sessionID)
+	if err != nil {
+		return nil
+	}
+	return &sid
+}