oikos/internal/httpapi/approvals.go

package httpapi

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"time"

	"github.com/dtoro/oikos/internal/db/sqlcgen"
	"github.com/dtoro/oikos/internal/domain"
	"github.com/dtoro/oikos/internal/httpapi/gen"
	"github.com/dtoro/oikos/internal/observability"
	"github.com/dtoro/oikos/internal/safego"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

// ─── Approvals ─────────────────────────────────────────────────────────

func (s *Server) ListApprovals(ctx context.Context, req gen.ListApprovalsRequestObject) (gen.ListApprovalsResponseObject, error) {
	limit := clampLimit(req.Params.Limit)
	var status *string
	if req.Params.Status != nil {
		s := string(*req.Params.Status)
		status = &s
	}
	var kind *string
	if req.Params.Kind != nil {
		k := string(*req.Params.Kind)
		kind = &k
	}
	rows, err := s.pool.Query(ctx, `
		SELECT a.entity_id, a.action, a.risk_class, a.kind, a.payload,
		       a.status, a.expires_at, a.decided_at, a.decided_by::text,
		       a.created_at, e.slug
		FROM approvals a
		JOIN entities e ON e.id = COALESCE(a.subject_entity_id, a.entity_id)
		WHERE ($1::text IS NULL OR a.status = $1)
		  AND ($2::text IS NULL OR a.kind = $2)
		  AND ($3::text IS NULL OR e.slug > $3)
		ORDER BY e.slug
		LIMIT $4`,
		status, kind, req.Params.Cursor, limit+1)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.Approval{}
	for rows.Next() {
		var a gen.Approval
		var payloadBytes []byte
		var decidedBy *string
		if err := rows.Scan(&a.Id, &a.Action, &a.RiskClass, &a.Kind, &payloadBytes,
			&a.Status, &a.ExpiresAt, &a.DecidedAt, &decidedBy,
			&a.CreatedAt, &a.Slug); err != nil {
			return nil, err
		}
		a.DecidedBy = decidedBy
		var payload map[string]any
		if len(payloadBytes) > 0 && json.Unmarshal(payloadBytes, &payload) == nil {
			a.Payload = &payload
		}
		items = append(items, a)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}

	var next *string
	if len(items) > limit {
		items = items[:limit]
		next = &items[len(items)-1].Slug
	}
	if items == nil {
		items = []gen.Approval{}
	}
	return gen.ListApprovals200JSONResponse{Items: items, NextCursor: next}, nil
}

func (s *Server) DecideApproval(ctx context.Context, req gen.DecideApprovalRequestObject) (gen.DecideApprovalResponseObject, error) {
	if req.Body == nil {
		return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput)
	}

	id, err := s.resolveEntityID(ctx, req.Id)
	if err != nil {
		return nil, err
	}

	actorType, actor := actorInfo(ctx)

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)

	q := sqlcgen.New(tx)

	// Verify HMAC token if provided (single-use, S5).
	if req.Body.Token != nil && *req.Body.Token != "" {
		var tokenHash *string
		var apprStatus string
		var expiresAt time.Time
		err := tx.QueryRow(ctx,
			"SELECT token_hash, status, expires_at FROM approvals WHERE entity_id = $1",
			id).Scan(&tokenHash, &apprStatus, &expiresAt)
		if err != nil || tokenHash == nil {
			return nil, fmt.Errorf("%w: approval not found", domain.ErrNotFound)
		}
		if apprStatus != "pending" {
			return nil, fmt.Errorf("%w: approval already decided", domain.ErrInvalidTransition)
		}
		if expiresAt.Before(time.Now()) {
			return nil, fmt.Errorf("%w: approval token expired", domain.ErrInvalidTransition)
		}
		if *tokenHash != hashToken(*req.Body.Token) {
			return nil, fmt.Errorf("%w: invalid approval token", domain.ErrInvalidInput)
		}
	}

	// Map decision to status.
	var status string
	switch req.Body.Decision {
	case gen.Approve:
		status = "approved"
	case gen.Deny:
		status = "denied"
	case gen.Revoke:
		status = "revoked"
	default:
		return nil, fmt.Errorf("%w: invalid decision %q", domain.ErrInvalidInput, req.Body.Decision)
	}

	if err := q.UpdateApprovalStatus(ctx, sqlcgen.UpdateApprovalStatusParams{
		EntityID: id,
		Status:   status,
	}); err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: approval %s not found or already decided", domain.ErrNotFound, req.Id)
		}
		return nil, err
	}

	// Re-read approval.
	app, err := q.GetApprovalByID(ctx, id)
	if err != nil {
		return nil, err
	}

	approval := approvalToGen(app)

	if auditErr := observability.Audit(ctx, q, actorType, actor, "decide",
		&id, "POST", "/api/v1/approvals/"+req.Id+"/decision", "",
		nil,
		map[string]any{"decision": status}); auditErr != nil {
		return nil, auditErr
	}

	// Emit for SSE fan-out (in-tx; NOTIFY fires post-commit).
	if evErr := observability.Event(ctx, q, "approval.decided", &id, "info", "api", "",
		map[string]any{"decision": status, "actor": actor}); evErr != nil {
		return nil, evErr
	}

	// On approve: execute the linked gated command.
	if status == "approved" {
		var execID, targetID uuid.UUID
		var actionStr, targetSlug, riskClass string
		err := tx.QueryRow(ctx, `
			SELECT e.entity_id, e.target_entity_id, e.action, e.risk_class
			FROM executions e
			WHERE e.approval_id = $1 AND e.status = 'pending_approval'
			LIMIT 1`, id).Scan(&execID, &targetID, &actionStr, &riskClass)
		if err == nil {
			// Resolve target entity slug from targetID.
			_ = tx.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", targetID).Scan(&targetSlug)

			safego.Go("httpapi:executeApprovedAction", func() {
				executeApprovedAction(context.Background(), s.pool, execID, targetSlug, actionStr)
			})
			// Status only — risk_class was set correctly at request time
			// (e.g. by policy.ClassifyCommand for `run`); overwriting it to
			// a hardcoded 'config_mutation' here corrupted the audit ledger
			// for every other risk class, including destructive.
			_, _ = tx.Exec(ctx, `UPDATE executions SET status = 'approved' WHERE entity_id = $1`, execID)

			// Approving a plan step — by ANY route (this endpoint backs both
			// the chat Approve button and chat-assent) — opens/extends the
			// agent's assent window. This is the scope gate the Nomos
			// auto-continuation worker checks: with the window open, the
			// finished execution's result is fed back to the agent so it runs
			// the plan to completion. Without opening it here, approving via
			// the button (instead of typing "go ahead") would silently not
			// auto-continue.
			var agentID *uuid.UUID
			if qerr := tx.QueryRow(ctx, "SELECT agent_id FROM executions WHERE entity_id = $1", execID).Scan(&agentID); qerr == nil && agentID != nil {
				expires := time.Now().Add(30 * time.Minute).UTC().Format(time.RFC3339)
				_, _ = tx.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2)
					ON CONFLICT (key) DO UPDATE SET value = $2`, "assent_window.agent:"+agentID.String(), expires)

				// Approving a DESTRUCTIVE step via the button is exactly as
				// explicit as a typed "I confirm" — the operator affirmatively
				// clicked Approve on a card that said DESTRUCTIVE. Open the
				// same short, target-scoped destructive window chat-assent's
				// typed-confirm path opens, for parity: a multi-step
				// destructive recovery (stop, then destroy) shouldn't need a
				// fresh confirmation per click any more than it needs one per
				// typed phrase.
				if riskClass == "destructive" && targetSlug != "" {
					dExpires := time.Now().Add(15 * time.Minute).UTC().Format(time.RFC3339)
					_, _ = tx.Exec(ctx, `INSERT INTO autonomy_settings (key, value) VALUES ($1, $2)
						ON CONFLICT (key) DO UPDATE SET value = $2`,
						"destructive_window.agent:"+agentID.String()+".target:"+targetSlug, dExpires)
				}
			}

			slog.Info("httpapi: approved execution queued",
				"execution_id", execID, "target", targetSlug, "action", actionStr)
		} else {
			slog.Warn("httpapi: no pending execution found for approval", "approval_id", id, "error", err)
		}
	} else {
		// Denied/revoked: reflect it on the linked execution too. Previously
		// only the approvals row changed, so the execution stayed
		// 'pending_approval' forever — any UI/poller reading execution
		// status (not approval status) never saw the decision.
		_, _ = tx.Exec(ctx, `UPDATE executions SET status = $2, completed_at = now() WHERE approval_id = $1 AND status = 'pending_approval'`, id, status)
	}

	// If this execution belongs to a nomos session, flip it out of
	// awaiting_input — the counterpart to classifyAndGate flipping it IN
	// the moment the approval was created (internal/mcp/server.go's
	// markSessionAwaitingApproval). Runs for all three decisions (approve/
	// deny/revoke): each one is an operator answer to "what do I do about
	// this?", same as answerQuestion's unconditional resume-to-executing
	// (cmd/nomos/store.go) for a session_questions answer.
	var awaitingSessionID string
	_ = tx.QueryRow(ctx, `
		SELECT pe.session_id FROM nomos_plan_executions pe
		JOIN executions ex ON ex.entity_id = pe.execution_id
		WHERE ex.approval_id = $1
		LIMIT 1`, id).Scan(&awaitingSessionID)
	if awaitingSessionID != "" {
		if rtag, rerr := tx.Exec(ctx, `
			UPDATE agent_sessions SET status = 'executing', last_active_at = now()
			WHERE id = $1 AND status = 'awaiting_input'`, awaitingSessionID); rerr == nil && rtag.RowsAffected() > 0 {
			var taskEntID *uuid.UUID
			var e uuid.UUID
			if qerr := tx.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, awaitingSessionID).Scan(&e); qerr == nil && e != uuid.Nil {
				taskEntID = &e
			}
			_ = observability.Event(ctx, q, "task.status", taskEntID, "info", "api", awaitingSessionID,
				map[string]any{"status": "executing", "reason": "approval_decided", "decision": status})
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}

	return gen.DecideApproval200JSONResponse(approval), nil
}

func approvalToGen(a sqlcgen.Approval) gen.Approval {
	app := gen.Approval{
		Id:        a.EntityID,
		Action:    a.Action,
		RiskClass: a.RiskClass,
		Kind:      gen.ApprovalKind(a.Kind),
		Status:    gen.ApprovalStatus(a.Status),
		ExpiresAt: a.ExpiresAt,
		DecidedAt: a.DecidedAt,
		CreatedAt: a.CreatedAt,
	}
	if a.DecidedBy != nil {
		s := a.DecidedBy.String()
		app.DecidedBy = &s
	}
	var payload map[string]any
	if len(a.Payload) > 0 && json.Unmarshal(a.Payload, &payload) == nil && len(payload) > 0 {
		app.Payload = &payload
	}
	return app
}