oikos/internal/httpapi/phase3.go

package httpapi

import (
	"context"
	"encoding/json"
	"fmt"
	"log/slog"
	"strings"
	"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/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgtype"
)

// ─── Checks ────────────────────────────────────────────────────────────

func (s *Server) ListChecks(ctx context.Context, req gen.ListChecksRequestObject) (gen.ListChecksResponseObject, error) {
	limit := clampLimit(req.Params.Limit)
	rows, err := s.pool.Query(ctx, `
		SELECT cd.entity_id, e.slug, cd.kind,
		       COALESCE(te.slug, '') AS target_slug, cd.target_type,
		       cd.config, cd.interval_s, cd.timeout_s, cd.zone, cd.enabled,
		       e.version
		FROM check_defs cd
		JOIN entities e ON e.id = cd.entity_id
		LEFT JOIN entities te ON te.id = cd.target_id
		WHERE ($1::text IS NULL OR cd.kind = $1)
		  AND ($2::text IS NULL OR te.slug = $2)
		  AND ($3::bool IS NULL OR cd.enabled = $3)
		  AND ($4::text IS NULL OR e.slug > $4)
		ORDER BY e.slug
		LIMIT $5`,
		req.Params.Kind, req.Params.Target, req.Params.Enabled, req.Params.Cursor, limit+1)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.Check{}
	for rows.Next() {
		var c gen.Check
		var targetSlug string
		var configBytes []byte
		if err := rows.Scan(&c.Id, &c.Slug, &c.Kind, &targetSlug, &c.TargetType,
			&configBytes, &c.IntervalS, &c.TimeoutS, &c.Zone, &c.Enabled, &c.Version); err != nil {
			return nil, err
		}
		if targetSlug != "" {
			c.Target = &targetSlug
		}
		var config map[string]any
		if len(configBytes) > 0 && json.Unmarshal(configBytes, &config) == nil && len(config) > 0 {
			c.Config = &config
		}
		items = append(items, c)
	}
	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.Check{}
	}
	return gen.ListChecks200JSONResponse{Items: items, NextCursor: next}, nil
}

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

	id, err := uuid.NewV7()
	if err != nil {
		return nil, err
	}

	slug := req.Body.Slug
	if slug == "" {
		slug = "check:" + string(req.Body.Kind) + ":" + uuid.New().String()[:8]
	}

	// Resolve target if provided.
	var targetID *uuid.UUID
	if req.Body.Target != nil && *req.Body.Target != "" {
		tid, rerr := s.resolveEntityID(ctx, *req.Body.Target)
		if rerr != nil {
			return nil, rerr
		}
		targetID = &tid
	}

	intervalS := int32(300)
	if req.Body.IntervalS != nil {
		intervalS = int32(*req.Body.IntervalS)
	}
	timeoutS := int32(30)
	if req.Body.TimeoutS != nil {
		timeoutS = int32(*req.Body.TimeoutS)
	}
	enabled := true
	if req.Body.Enabled != nil {
		enabled = *req.Body.Enabled
	}

	configJSON := []byte("{}")
	if req.Body.Config != nil {
		configJSON, _ = json.Marshal(req.Body.Config)
	}

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

	q := sqlcgen.New(tx)

	// Create the entity row (checks are entities).
	entity, err := q.InsertEntity(ctx, sqlcgen.InsertEntityParams{
		ID:         id,
		Slug:       slug,
		Type:       "check_def",
		Name:       slug,
		Attributes: []byte("{}"),
	})
	if err != nil {
		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
			return nil, fmt.Errorf("%w: check %q already exists", domain.ErrAlreadyExists, slug)
		}
		return nil, err
	}

	if err := q.InsertCheckDef(ctx, sqlcgen.InsertCheckDefParams{
		EntityID:   id,
		TargetID:   targetID,
		TargetType: req.Body.TargetType,
		Kind:       string(req.Body.Kind),
		Config:     configJSON,
		IntervalS:  intervalS,
		TimeoutS:   timeoutS,
		Zone:       req.Body.Zone,
		Enabled:    enabled,
	}); err != nil {
		return nil, err
	}

	// Build response Check.
	check := gen.Check{
		Id:         id,
		Slug:       entity.Slug,
		Kind:       gen.CheckKind(req.Body.Kind),
		IntervalS:  int(intervalS),
		TimeoutS:   int(timeoutS),
		Enabled:    enabled,
		TargetType: req.Body.TargetType,
		Zone:       req.Body.Zone,
		Version:    int(entity.Version),
	}
	if req.Body.Config != nil {
		check.Config = req.Body.Config
	}
	if targetID != nil && req.Body.Target != nil {
		check.Target = req.Body.Target
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, q, actorType, actor, "create",
		&id, "POST", "/api/v1/checks", "",
		map[string]any{"kind": req.Body.Kind, "slug": slug}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.CreateCheck201JSONResponse(check), nil
}

func (s *Server) PatchCheck(ctx context.Context, req gen.PatchCheckRequestObject) (gen.PatchCheckResponseObject, 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
	}

	// Parse If-Match
	ifMatch := strings.Trim(req.Params.IfMatch, `"`)
	expectedVersion, err := parseIntIfMatch(ifMatch)
	if err != nil {
		return nil, err
	}
	_ = expectedVersion // check_defs don't track version via If-Match today, but we validate the header is present

	if ifMatch == "" {
		return nil, fmt.Errorf("%w: invalid If-Match header", domain.ErrInvalidInput)
	}

	// Get current check def
	current, err := sqlcgen.New(s.pool).GetCheckDef(ctx, id)
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: check %s", domain.ErrNotFound, req.Id)
		}
		return nil, err
	}

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

	// Apply patch.
	if req.Body.Config != nil {
		current.Config, _ = json.Marshal(req.Body.Config)
	}
	if req.Body.IntervalS != nil {
		current.IntervalS = int32(*req.Body.IntervalS)
	}
	if req.Body.TimeoutS != nil {
		current.TimeoutS = int32(*req.Body.TimeoutS)
	}
	if req.Body.Enabled != nil {
		current.Enabled = *req.Body.Enabled
	}

	if err := sqlcgen.New(tx).UpdateCheckDef(ctx, sqlcgen.UpdateCheckDefParams{
		EntityID:   id,
		Kind:       current.Kind,
		Config:     current.Config,
		IntervalS:  current.IntervalS,
		TimeoutS:   current.TimeoutS,
		TargetID:   current.TargetID,
		TargetType: current.TargetType,
		Zone:       current.Zone,
		Enabled:    current.Enabled,
	}); err != nil {
		return nil, err
	}

	// Re-read to get updated timestamp.
	updated, err := sqlcgen.New(tx).GetCheckDef(ctx, id)
	if err != nil {
		return nil, err
	}

	check := checkDefToGen(updated)

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "patch",
		&id, "PATCH", "/api/v1/checks/"+req.Id, "",
		map[string]any{"enabled": updated.Enabled}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.PatchCheck200JSONResponse(check), nil
}

func checkDefToGen(cd sqlcgen.CheckDef) gen.Check {
	c := gen.Check{
		Id:         cd.EntityID,
		Kind:       gen.CheckKind(cd.Kind),
		IntervalS:  int(cd.IntervalS),
		TimeoutS:   int(cd.TimeoutS),
		Enabled:    cd.Enabled,
		TargetType: cd.TargetType,
		Zone:       cd.Zone,
	}
	var config map[string]any
	if len(cd.Config) > 0 && json.Unmarshal(cd.Config, &config) == nil && len(config) > 0 {
		c.Config = &config
	}
	return c
}

// parseIntIfMatch parses an integer from a raw If-Match header value (with quotes stripped).
func parseIntIfMatch(s string) (int, error) {
	if s == "" {
		return 0, fmt.Errorf("empty version")
	}
	var v int
	for _, c := range s {
		if c < '0' || c > '9' {
			return 0, fmt.Errorf("invalid version: %q", s)
		}
		v = v*10 + int(c-'0')
	}
	return v, nil
}

// ─── Classifications ───────────────────────────────────────────────────

func (s *Server) ListClassifications(ctx context.Context, req gen.ListClassificationsRequestObject) (gen.ListClassificationsResponseObject, error) {
	limit := clampLimit(req.Params.Limit)
	var route *string
	if req.Params.Route != nil {
		r := string(*req.Params.Route)
		route = &r
	}
	rows, err := s.pool.Query(ctx, `
		SELECT c.entity_id, c.signal_entity_id, c.target_entity_id, c.action,
		       c.recommended_action, c.risk_class, c.route, c.blast_radius,
		       c.pattern_confidence, c.skill_id, c.autonomy_check, c.reasoning,
		       c.correlation_id, c.created_at,
		       e.slug, COALESCE(se.slug, '') AS signal_slug, COALESCE(te.slug, '') AS target_slug
		FROM classifications c
		LEFT JOIN entities e ON e.id = c.entity_id
		LEFT JOIN entities se ON se.id = c.signal_entity_id
		LEFT JOIN entities te ON te.id = c.target_entity_id
		WHERE ($1::text IS NULL OR c.route = $1)
		  AND ($2::text IS NULL OR e.slug > $2)
		ORDER BY e.slug
		LIMIT $3`,
		route, req.Params.Cursor, limit+1)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.Classification{}
	for rows.Next() {
		var cls gen.Classification
		var recActionJSON []byte
		var reasoningJSON []byte
		var blastRadius []uuid.UUID
		var signalSlug, targetSlug string
		if err := rows.Scan(&cls.Id, &cls.SignalId, &targetSlug, &cls.Action,
			&recActionJSON, &cls.RiskClass, &cls.Route, &blastRadius,
			&cls.PatternConfidence, &cls.SkillId, &cls.AutonomyCheck, &reasoningJSON,
			&cls.CorrelationId, &cls.CreatedAt,
			&cls.Target, &signalSlug, &targetSlug); err != nil {
			return nil, err
		}
		if targetSlug != "" {
			cls.Target = &targetSlug
		}
		var reasoning map[string]any
		if json.Unmarshal(reasoningJSON, &reasoning) == nil {
			cls.Reasoning = reasoning
		}
		if len(blastRadius) > 0 {
			br := make([]string, len(blastRadius))
			for i, id := range blastRadius {
				br[i] = id.String()
			}
			cls.BlastRadius = &br
		}
		items = append(items, cls)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}

	var next *string
	if len(items) > limit {
		items = items[:limit]
		if items[len(items)-1].Target != nil {
			next = items[len(items)-1].Target
		}
	}
	if items == nil {
		items = []gen.Classification{}
	}
	return gen.ListClassifications200JSONResponse{Items: items, NextCursor: next}, nil
}

// ─── Executions ────────────────────────────────────────────────────────

func (s *Server) ListExecutions(ctx context.Context, req gen.ListExecutionsRequestObject) (gen.ListExecutionsResponseObject, error) {
	limit := clampLimit(req.Params.Limit)
	rows, err := s.pool.Query(ctx, `
		SELECT e.entity_id, e.classification_id::text, e.signal_entity_id::text,
		       e.target_entity_id, e.action, e.risk_class,
		       e.approval_id::text, e.agent_id::text, e.skill_id::text,
		       e.skill_version, e.status, e.result, e.duration_ms,
		       e.verified, e.correlation_id, e.started_at, e.completed_at, e.created_at,
		       te.slug
		FROM executions e
		JOIN entities te ON te.id = e.target_entity_id
		WHERE ($1::text IS NULL OR e.status = $1)
		  AND ($2::text IS NULL OR te.slug > $2)
		ORDER BY te.slug
		LIMIT $3`,
		req.Params.Status, req.Params.Cursor, limit+1)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.Execution{}
	for rows.Next() {
		var exec gen.Execution
		var resultBytes []byte
		var targetSlug string
		if err := rows.Scan(&exec.Id, &exec.ClassificationId, &exec.SignalId,
			&exec.Target, &exec.Action, &exec.RiskClass,
			&exec.ApprovalId, &exec.AgentId, &exec.SkillId,
			&exec.SkillVersion, &exec.Status, &resultBytes, &exec.DurationMs,
			&exec.Verified, &exec.CorrelationId, &exec.StartedAt, &exec.CompletedAt,
			&exec.CreatedAt, &targetSlug); err != nil {
			return nil, err
		}
		var result map[string]any
		if len(resultBytes) > 0 && json.Unmarshal(resultBytes, &result) == nil {
			exec.Result = &result
		}
		// Target is stored as UUID, but we surface the slug
		exec.Slug = targetSlug
		items = append(items, exec)
	}
	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.Execution{}
	}
	return gen.ListExecutions200JSONResponse{Items: items, NextCursor: next}, nil
}

func (s *Server) GetExecution(ctx context.Context, req gen.GetExecutionRequestObject) (gen.GetExecutionResponseObject, error) {
	id, err := s.resolveEntityID(ctx, req.Id)
	if err != nil {
		return nil, err
	}

	var exec gen.Execution
	var resultBytes []byte
	var targetSlug string
	err = s.pool.QueryRow(ctx, `
		SELECT e.entity_id, e.classification_id::text, e.signal_entity_id::text,
		       e.target_entity_id, e.action, e.risk_class,
		       e.approval_id::text, e.agent_id::text, e.skill_id::text,
		       e.skill_version, e.status, e.result, e.duration_ms,
		       e.verified, e.correlation_id, e.started_at, e.completed_at, e.created_at,
		       te.slug
		FROM executions e
		JOIN entities te ON te.id = e.target_entity_id
		WHERE e.entity_id = $1`, id).
		Scan(&exec.Id, &exec.ClassificationId, &exec.SignalId,
			&exec.Target, &exec.Action, &exec.RiskClass,
			&exec.ApprovalId, &exec.AgentId, &exec.SkillId,
			&exec.SkillVersion, &exec.Status, &resultBytes, &exec.DurationMs,
			&exec.Verified, &exec.CorrelationId, &exec.StartedAt, &exec.CompletedAt,
			&exec.CreatedAt, &targetSlug)
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: execution %s", domain.ErrNotFound, req.Id)
		}
		return nil, err
	}
	var result map[string]any
	if len(resultBytes) > 0 && json.Unmarshal(resultBytes, &result) == nil {
		exec.Result = &result
	}
	exec.Slug = targetSlug
	return gen.GetExecution200JSONResponse(exec), nil
}

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

	id, err := uuid.NewV7()
	if err != nil {
		return nil, err
	}

	targetID, err := s.resolveEntityID(ctx, req.Body.Target)
	if err != nil {
		return nil, err
	}

	correlationID := uuid.New().String()

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

	q := sqlcgen.New(tx)

	execSlug := "exec:" + id.String()[:8]
	if _, err := q.InsertEntity(ctx, sqlcgen.InsertEntityParams{
		ID:         id,
		Slug:       execSlug,
		Type:       "execution",
		Name:       req.Body.Action + " on " + req.Body.Target,
		Attributes: []byte("{}"),
	}); err != nil {
		return nil, err
	}

	if err := q.InsertExecution(ctx, sqlcgen.InsertExecutionParams{
		EntityID:       id,
		TargetEntityID: &targetID,
		Action:         req.Body.Action,
		RiskClass:      "unclassified", // will be classified by classifier
		CorrelationID:  correlationID,
	}); err != nil {
		return nil, err
	}

	// Re-read to get the full record.
	var exec gen.Execution
	var resultBytes []byte
	var targetSlug string
	err = tx.QueryRow(ctx, `
		SELECT e.entity_id, e.classification_id::text, e.signal_entity_id::text,
		       e.target_entity_id, e.action, e.risk_class,
		       e.approval_id::text, e.agent_id::text, e.skill_id::text,
		       e.skill_version, e.status, e.result, e.duration_ms,
		       e.verified, e.correlation_id, e.started_at, e.completed_at, e.created_at,
		       te.slug
		FROM executions e
		JOIN entities te ON te.id = e.target_entity_id
		WHERE e.entity_id = $1`, id).
		Scan(&exec.Id, &exec.ClassificationId, &exec.SignalId,
			&exec.Target, &exec.Action, &exec.RiskClass,
			&exec.ApprovalId, &exec.AgentId, &exec.SkillId,
			&exec.SkillVersion, &exec.Status, &resultBytes, &exec.DurationMs,
			&exec.Verified, &exec.CorrelationId, &exec.StartedAt, &exec.CompletedAt,
			&exec.CreatedAt, &targetSlug)
	if err != nil {
		return nil, err
	}
	exec.Slug = targetSlug

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, q, actorType, actor, "create",
		&id, "POST", "/api/v1/executions", "",
		map[string]any{"action": req.Body.Action, "target": req.Body.Target}); auditErr != nil {
		return nil, auditErr
	}
	if eventErr := observability.Event(ctx, q, "execution.requested", &id,
		"info", "oikos-api", "",
		map[string]any{"action": req.Body.Action, "target": req.Body.Target}); eventErr != nil {
		return nil, eventErr
	}

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

	return gen.RequestExecution201JSONResponse(exec), nil
}

func (s *Server) CancelExecution(ctx context.Context, req gen.CancelExecutionRequestObject) (gen.CancelExecutionResponseObject, error) {
	id, err := s.resolveEntityID(ctx, req.Id)
	if err != nil {
		return nil, err
	}

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

	q := sqlcgen.New(tx)
	if err := q.UpdateExecutionStatus(ctx, sqlcgen.UpdateExecutionStatusParams{
		EntityID: id,
		Status:   "cancelled",
	}); err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: execution %s", domain.ErrNotFound, req.Id)
		}
		return nil, err
	}

	// Re-read.
	var exec gen.Execution
	var resultBytes []byte
	var targetSlug string
	err = tx.QueryRow(ctx, `
		SELECT e.entity_id, e.classification_id::text, e.signal_entity_id::text,
		       e.target_entity_id, e.action, e.risk_class,
		       e.approval_id::text, e.agent_id::text, e.skill_id::text,
		       e.skill_version, e.status, e.result, e.duration_ms,
		       e.verified, e.correlation_id, e.started_at, e.completed_at, e.created_at,
		       te.slug
		FROM executions e
		JOIN entities te ON te.id = e.target_entity_id
		WHERE e.entity_id = $1`, id).
		Scan(&exec.Id, &exec.ClassificationId, &exec.SignalId,
			&exec.Target, &exec.Action, &exec.RiskClass,
			&exec.ApprovalId, &exec.AgentId, &exec.SkillId,
			&exec.SkillVersion, &exec.Status, &resultBytes, &exec.DurationMs,
			&exec.Verified, &exec.CorrelationId, &exec.StartedAt, &exec.CompletedAt,
			&exec.CreatedAt, &targetSlug)
	if err != nil {
		return nil, err
	}
	exec.Slug = targetSlug

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

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

	return gen.CancelExecution200JSONResponse(exec), nil
}

// ─── 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)

	// 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", "",
		map[string]any{"decision": status}); auditErr != nil {
		return nil, auditErr
	}

	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
}

// ─── Patterns ──────────────────────────────────────────────────────────

func (s *Server) ListPatterns(ctx context.Context, req gen.ListPatternsRequestObject) (gen.ListPatternsResponseObject, error) {
	limit := clampLimit(req.Params.Limit)
	rows, err := s.pool.Query(ctx, `
		SELECT p.entity_id, e.slug, p.applies_type, p.action, p.pattern, p.confidence,
		       p.evidence_count, p.success_count, p.failure_count, p.status,
		       p.quarantined, p.version, p.last_validated_at
		FROM patterns p
		JOIN entities e ON e.id = p.entity_id
		WHERE ($1::text IS NULL OR p.status = $1)
		  AND ($2::text IS NULL OR p.applies_type = $2)
		  AND ($3::text IS NULL OR p.action = $3)
		ORDER BY p.applies_type, p.action
		LIMIT $4`,
		req.Params.Status, req.Params.EntityType, req.Params.Action, limit+1)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.Pattern{}
	for rows.Next() {
		var p gen.Pattern
		if err := rows.Scan(&p.Id, &p.Slug, &p.AppliesType, &p.Action, &p.Pattern,
			&p.Confidence, &p.EvidenceCount, &p.SuccessCount, &p.FailureCount,
			&p.Status, &p.Quarantined, &p.Version, &p.LastValidatedAt); err != nil {
			return nil, err
		}
		items = append(items, p)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}

	if items == nil {
		items = []gen.Pattern{}
	}
	return gen.ListPatterns200JSONResponse{Items: items}, nil
}

func (s *Server) PatchPattern(ctx context.Context, req gen.PatchPatternRequestObject) (gen.PatchPatternResponseObject, 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
	}

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

	if req.Body.Status != nil {
		status := string(*req.Body.Status)
		if err := q.UpdatePatternStatus(ctx, sqlcgen.UpdatePatternStatusParams{
			EntityID: id,
			Status:   status,
		}); err != nil {
			if err == pgx.ErrNoRows {
				return nil, fmt.Errorf("%w: pattern %s", domain.ErrNotFound, req.Id)
			}
			return nil, err
		}
	}

	if req.Body.Quarantined != nil {
		if err := q.UpdatePatternQuarantine(ctx, sqlcgen.UpdatePatternQuarantineParams{
			EntityID:    id,
			Quarantined: *req.Body.Quarantined,
		}); err != nil {
			return nil, err
		}
	}

	// Re-read.
	var p gen.Pattern
	err = tx.QueryRow(ctx, `
		SELECT entity_id, applies_type, action, pattern, confidence,
		       evidence_count, success_count, failure_count, status,
		       quarantined, version, last_validated_at
		FROM patterns WHERE entity_id = $1`, id).
		Scan(&p.Id, &p.AppliesType, &p.Action, &p.Pattern,
			&p.Confidence, &p.EvidenceCount, &p.SuccessCount, &p.FailureCount,
			&p.Status, &p.Quarantined, &p.Version, &p.LastValidatedAt)
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: pattern %s", domain.ErrNotFound, req.Id)
		}
		return nil, err
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, q, actorType, actor, "patch",
		&id, "PATCH", "/api/v1/patterns/"+req.Id, "",
		map[string]any{"status": req.Body.Status, "quarantined": req.Body.Quarantined}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.PatchPattern200JSONResponse(p), nil
}

// ─── Skills ────────────────────────────────────────────────────────────

func (s *Server) ListSkills(ctx context.Context, req gen.ListSkillsRequestObject) (gen.ListSkillsResponseObject, error) {
	limit := clampLimit(req.Params.Limit)
	rows, err := s.pool.Query(ctx, `
		SELECT s.entity_id, s.version, s.name, s.procedure, s.applies_type,
		       s.action, s.pattern_ids, s.status, s.success_rate,
		       s.changed_by::text, s.change_reason, s.last_used_at
		FROM skills s
		WHERE ($1::text IS NULL OR s.status = $1)
		  AND ($2::text IS NULL OR s.applies_type = $2)
		  AND ($3::text IS NULL OR s.action = $3)
		ORDER BY s.name, s.version DESC`,
		req.Params.Status, req.Params.AppliesTo, req.Params.Action)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	// Deduplicate to latest version per skill (the ORDER BY name, version DESC
	// means the first row per name is the latest).
	seen := map[string]bool{}
	items := []gen.Skill{}
	for rows.Next() {
		var s gen.Skill
		var procBytes []byte
		var patternIDs []uuid.UUID
		if err := rows.Scan(&s.Id, &s.Version, &s.Name, &procBytes, &s.AppliesType,
			&s.Action, &patternIDs, &s.Status, &s.SuccessRate,
			&s.ChangedBy, &s.ChangeReason, &s.LastUsedAt); err != nil {
			return nil, err
		}
		if seen[s.Id.String()] {
			continue
		}
		seen[s.Id.String()] = true
		if err := json.Unmarshal(procBytes, &s.Procedure); err != nil {
			slog.Warn("phase3: unmarshal skill procedure", "skill", s.Name, "error", err)
		}
		if len(patternIDs) > 0 {
			pids := make([]string, len(patternIDs))
			for i, pid := range patternIDs {
				pids[i] = pid.String()
			}
			s.PatternIds = &pids
		}
		items = append(items, s)
		if len(items) > limit {
			break
		}
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}

	if items == nil {
		items = []gen.Skill{}
	}
	return gen.ListSkills200JSONResponse{Items: items}, nil
}

func (s *Server) PatchSkill(ctx context.Context, req gen.PatchSkillRequestObject) (gen.PatchSkillResponseObject, 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
	}

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

	if req.Body.Status != nil {
		if err := q.UpdateSkillStatus(ctx, sqlcgen.UpdateSkillStatusParams{
			EntityID: id,
			Status:   string(*req.Body.Status),
		}); err != nil {
			if err == pgx.ErrNoRows {
				return nil, fmt.Errorf("%w: skill %s", domain.ErrNotFound, req.Id)
			}
			return nil, err
		}
	}

	// Re-read skill.
	var skill gen.Skill
	var procBytes []byte
	var patternIDs []uuid.UUID
	err = tx.QueryRow(ctx, `
		SELECT entity_id, version, name, procedure, applies_type, action,
		       pattern_ids, status, success_rate, changed_by::text,
		       change_reason, last_used_at
		FROM skills WHERE entity_id = $1 ORDER BY version DESC LIMIT 1`, id).
		Scan(&skill.Id, &skill.Version, &skill.Name, &procBytes, &skill.AppliesType,
			&skill.Action, &patternIDs, &skill.Status, &skill.SuccessRate,
			&skill.ChangedBy, &skill.ChangeReason, &skill.LastUsedAt)
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: skill %s", domain.ErrNotFound, req.Id)
		}
		return nil, err
	}
	if err := json.Unmarshal(procBytes, &skill.Procedure); err != nil {
			slog.Warn("phase3: unmarshal skill proc", "error", err)
		}
	if len(patternIDs) > 0 {
		pids := make([]string, len(patternIDs))
		for i, pid := range patternIDs {
			pids[i] = pid.String()
		}
		skill.PatternIds = &pids
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, q, actorType, actor, "patch",
		&id, "PATCH", "/api/v1/skills/"+req.Id, "",
		map[string]any{"status": req.Body.Status, "pinned_version": req.Body.PinnedVersion}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.PatchSkill200JSONResponse(skill), nil
}

func (s *Server) ListSkillVersions(ctx context.Context, req gen.ListSkillVersionsRequestObject) (gen.ListSkillVersionsResponseObject, error) {
	id, err := s.resolveEntityID(ctx, req.Id)
	if err != nil {
		return nil, err
	}

	rows, err := s.pool.Query(ctx, `
		SELECT entity_id, version, name, procedure, applies_type, action,
		       pattern_ids, status, success_rate, changed_by::text,
		       change_reason, last_used_at
		FROM skills WHERE entity_id = $1
		ORDER BY version DESC`, id)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.Skill{}
	for rows.Next() {
		var skill gen.Skill
		var procBytes []byte
		var patternIDs []uuid.UUID
		if err := rows.Scan(&skill.Id, &skill.Version, &skill.Name, &procBytes, &skill.AppliesType,
			&skill.Action, &patternIDs, &skill.Status, &skill.SuccessRate,
			&skill.ChangedBy, &skill.ChangeReason, &skill.LastUsedAt); err != nil {
			return nil, err
		}
		if err := json.Unmarshal(procBytes, &skill.Procedure); err != nil {
			slog.Warn("phase3: unmarshal skill proc", "error", err)
		}
		if len(patternIDs) > 0 {
			pids := make([]string, len(patternIDs))
			for i, pid := range patternIDs {
				pids[i] = pid.String()
			}
			skill.PatternIds = &pids
		}
		items = append(items, skill)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}
	if items == nil {
		items = []gen.Skill{}
	}
	return gen.ListSkillVersions200JSONResponse{Items: items}, nil
}

// ─── Approval Rules (Policy) ───────────────────────────────────────────

func (s *Server) ListApprovalRules(ctx context.Context, req gen.ListApprovalRulesRequestObject) (gen.ListApprovalRulesResponseObject, error) {
	rows, err := s.pool.Query(ctx, `
		SELECT id, entity_type, action, risk_class, autonomy_level,
		       COALESCE((SELECT slug FROM entities WHERE id = scope_entity), ''),
		       version, updated_at
		FROM approval_rules ORDER BY entity_type, action`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.ApprovalRule{}
	for rows.Next() {
		var rule gen.ApprovalRule
		var scopeSlug string
		if err := rows.Scan(&rule.Id, &rule.EntityType, &rule.Action,
			&rule.RiskClass, &rule.AutonomyLevel, &scopeSlug,
			&rule.Version); err != nil {
			return nil, err
		}
		if scopeSlug != "" {
			rule.ScopeEntity = &scopeSlug
		}
		items = append(items, rule)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}
	if items == nil {
		items = []gen.ApprovalRule{}
	}
	return gen.ListApprovalRules200JSONResponse{Items: items}, nil
}

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

	id, err := uuid.NewV7()
	if err != nil {
		return nil, err
	}

	var scopeEntity *uuid.UUID
	if req.Body.ScopeEntity != nil && *req.Body.ScopeEntity != "" {
		se, rerr := s.resolveEntityID(ctx, *req.Body.ScopeEntity)
		if rerr != nil {
			return nil, rerr
		}
		scopeEntity = &se
	}

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

	_, err = tx.Exec(ctx, `
		INSERT INTO approval_rules (id, entity_type, action, risk_class, autonomy_level, scope_entity)
		VALUES ($1, $2, $3, $4, $5, $6)`,
		id, req.Body.EntityType, req.Body.Action, req.Body.RiskClass,
		string(req.Body.AutonomyLevel), scopeEntity)
	if err != nil {
		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
			return nil, fmt.Errorf("%w: rule for %s/%s already exists", domain.ErrAlreadyExists,
				coalesceStr(req.Body.EntityType, "*"), req.Body.Action)
		}
		return nil, err
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "create",
		&id, "POST", "/api/v1/policy/approval-rules", "",
		map[string]any{"action": req.Body.Action, "risk_class": req.Body.RiskClass}); auditErr != nil {
		return nil, auditErr
	}

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

	// Return 202 pending approval (dual-control).
	return gen.CreateApprovalRule202JSONResponse{}, nil
}

func (s *Server) PatchApprovalRule(ctx context.Context, req gen.PatchApprovalRuleRequestObject) (gen.PatchApprovalRuleResponseObject, 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
	}

	var scopeEntity *uuid.UUID
	if req.Body.ScopeEntity != nil && *req.Body.ScopeEntity != "" {
		se, rerr := s.resolveEntityID(ctx, *req.Body.ScopeEntity)
		if rerr != nil {
			return nil, rerr
		}
		scopeEntity = &se
	}

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

	result, err := tx.Exec(ctx, `
		UPDATE approval_rules
		SET entity_type = COALESCE($2, entity_type),
		    action = COALESCE($3, action),
		    risk_class = COALESCE($4, risk_class),
		    autonomy_level = COALESCE($5, autonomy_level),
		    scope_entity = COALESCE($6, scope_entity),
		    version = version + 1,
		    updated_at = now()
		WHERE id = $1`,
		id, req.Body.EntityType, req.Body.Action, req.Body.RiskClass,
		string(req.Body.AutonomyLevel), scopeEntity)
	if err != nil {
		return nil, err
	}
	if result.RowsAffected() == 0 {
		return nil, fmt.Errorf("%w: approval rule %s", domain.ErrNotFound, req.Id)
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "patch",
		&id, "PATCH", "/api/v1/policy/approval-rules/"+req.Id, "",
		map[string]any{"action": req.Body.Action}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.PatchApprovalRule202JSONResponse{}, nil
}

// ─── Autonomy Settings ─────────────────────────────────────────────────

func (s *Server) GetAutonomySettings(ctx context.Context, req gen.GetAutonomySettingsRequestObject) (gen.GetAutonomySettingsResponseObject, error) {
	rows, err := s.pool.Query(ctx, `SELECT key, value, version, updated_at FROM autonomy_settings ORDER BY key`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.AutonomySetting{}
	for rows.Next() {
		var as gen.AutonomySetting
		if err := rows.Scan(&as.Key, &as.Value, &as.Version, &as.UpdatedAt); err != nil {
			return nil, err
		}
		items = append(items, as)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}
	if items == nil {
		items = []gen.AutonomySetting{}
	}
	return gen.GetAutonomySettings200JSONResponse{Items: items}, nil
}

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

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

	for key, value := range *req.Body {
		_, err := tx.Exec(ctx, `
			INSERT INTO autonomy_settings (key, value, version, updated_at)
			VALUES ($1, $2, 1, now())
			ON CONFLICT (key)
			DO UPDATE SET value = EXCLUDED.value, version = autonomy_settings.version + 1, updated_at = now()`,
			key, value)
		if err != nil {
			return nil, err
		}
	}

	// Re-read all settings.
	rows, err := tx.Query(ctx, `SELECT key, value, version, updated_at FROM autonomy_settings ORDER BY key`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.AutonomySetting{}
	for rows.Next() {
		var as gen.AutonomySetting
		if err := rows.Scan(&as.Key, &as.Value, &as.Version, &as.UpdatedAt); err != nil {
			return nil, err
		}
		items = append(items, as)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "patch",
		nil, "PATCH", "/api/v1/policy/autonomy", "",
		map[string]any{"keys": keysOfMap(*req.Body)}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.PatchAutonomySettings200JSONResponse{Items: items}, nil
}

// keysOfMap returns the keys of a map[string]string.
func keysOfMap(m map[string]string) []string {
	keys := make([]string, 0, len(m))
	for k := range m {
		keys = append(keys, k)
	}
	return keys
}

// ─── Risk Classes ──────────────────────────────────────────────────────

func (s *Server) ListRiskClasses(ctx context.Context, req gen.ListRiskClassesRequestObject) (gen.ListRiskClassesResponseObject, error) {
	rows, err := s.pool.Query(ctx, `SELECT name, description, approval_required, autonomy_allowed FROM risk_classes ORDER BY name`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.RiskClass{}
	for rows.Next() {
		var rc gen.RiskClass
		if err := rows.Scan(&rc.Name, &rc.Description, &rc.ApprovalRequired, &rc.AutonomyAllowed); err != nil {
			return nil, err
		}
		items = append(items, rc)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}
	if items == nil {
		items = []gen.RiskClass{}
	}
	return gen.ListRiskClasses200JSONResponse{Items: items}, nil
}

// ─── Relationships ─────────────────────────────────────────────────────

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

	sourceID, err := s.resolveEntityID(ctx, req.Body.Source)
	if err != nil {
		return nil, err
	}
	targetID, err := s.resolveEntityID(ctx, req.Body.Target)
	if err != nil {
		return nil, err
	}

	attrsJSON := []byte("{}")
	if req.Body.Attributes != nil {
		attrsJSON, _ = json.Marshal(req.Body.Attributes)
	}

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

	_, err = tx.Exec(ctx, `
		INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
		VALUES ($1, $2, $3, $4, now())`,
		sourceID, targetID, req.Body.Type, attrsJSON)
	if err != nil {
		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
			return nil, fmt.Errorf("%w: relationship %s:%s:%s already exists",
				domain.ErrAlreadyExists, req.Body.Source, req.Body.Type, req.Body.Target)
		}
		return nil, err
	}

	rel := gen.Relationship{
		Source:    req.Body.Source,
		Target:    req.Body.Target,
		Type:      req.Body.Type,
		ValidFrom: time.Now(),
	}
	if req.Body.Attributes != nil {
		rel.Attributes = req.Body.Attributes
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "create",
		nil, "POST", "/api/v1/relationships", "",
		map[string]any{"source": req.Body.Source, "target": req.Body.Target, "type": req.Body.Type}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.CreateRelationship201JSONResponse(rel), nil
}

func (s *Server) EndRelationship(ctx context.Context, req gen.EndRelationshipRequestObject) (gen.EndRelationshipResponseObject, error) {
	sourceID, err := s.resolveEntityID(ctx, req.Params.Source)
	if err != nil {
		return nil, err
	}
	targetID, err := s.resolveEntityID(ctx, req.Params.Target)
	if err != nil {
		return nil, err
	}

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

	result, err := tx.Exec(ctx, `
		UPDATE relationships
		SET valid_to = now()
		WHERE source_id = $1 AND target_id = $2 AND type = $3 AND valid_to IS NULL`,
		sourceID, targetID, req.Params.RelType)
	if err != nil {
		return nil, err
	}
	if result.RowsAffected() == 0 {
		return nil, fmt.Errorf("%w: active relationship %s:%s:%s",
			domain.ErrNotFound, req.Params.Source, req.Params.RelType, req.Params.Target)
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "delete",
		nil, "DELETE", "/api/v1/relationships", "",
		map[string]any{"source": req.Params.Source, "target": req.Params.Target, "type": req.Params.RelType}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.EndRelationship204Response{}, nil
}

// ─── Entity Types (Ontology) ───────────────────────────────────────────

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

	isAbstract := false
	if req.Body.IsAbstract != nil {
		isAbstract = *req.Body.IsAbstract
	}

	attrsSchemaJSON := []byte("null")
	if req.Body.AttributeSchema != nil {
		attrsSchemaJSON, _ = json.Marshal(req.Body.AttributeSchema)
	}

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

	_, err = tx.Exec(ctx, `
		INSERT INTO entity_types (name, parent_type, is_abstract, domain, layer, description, lifecycle_id, attribute_schema, status)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 'active')`,
		req.Body.Name, req.Body.ParentType, isAbstract, req.Body.Domain,
		string(req.Body.Layer), req.Body.Description, req.Body.LifecycleId, attrsSchemaJSON)
	if err != nil {
		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
			return nil, fmt.Errorf("%w: entity type %q already exists", domain.ErrAlreadyExists, req.Body.Name)
		}
		return nil, err
	}

	// Re-read.
	var et gen.EntityType
	var schemaBytes []byte
	err = tx.QueryRow(ctx, `
		SELECT name, parent_type, is_abstract, domain, layer, description,
		       lifecycle_id, attribute_schema, schema_version, status
		FROM entity_types WHERE name = $1`, req.Body.Name).
		Scan(&et.Name, &et.ParentType, &et.IsAbstract, &et.Domain, &et.Layer,
			&et.Description, &et.LifecycleId, &schemaBytes, &et.SchemaVersion, &et.Status)
	if err != nil {
		return nil, err
	}
	var schema map[string]any
	if len(schemaBytes) > 0 && json.Unmarshal(schemaBytes, &schema) == nil && schema != nil {
		et.AttributeSchema = &schema
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "create",
		nil, "POST", "/api/v1/ontology/entity-types", "",
		map[string]any{"name": req.Body.Name, "domain": req.Body.Domain}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.CreateEntityType201JSONResponse(et), nil
}

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

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

	// Build dynamic update.
	sets := []string{}
	args := []any{}
	argIdx := 2

	if req.Body.Description != nil {
		sets = append(sets, fmt.Sprintf("description = $%d", argIdx))
		args = append(args, *req.Body.Description)
		argIdx++
	}
	if req.Body.Status != nil {
		sets = append(sets, fmt.Sprintf("status = $%d", argIdx))
		args = append(args, string(*req.Body.Status))
		argIdx++
	}
	if req.Body.AttributeSchema != nil {
		schemaJSON, _ := json.Marshal(req.Body.AttributeSchema)
		sets = append(sets, fmt.Sprintf("attribute_schema = $%d", argIdx))
		args = append(args, schemaJSON)
		argIdx++
	}

	if len(sets) == 0 {
		return nil, fmt.Errorf("%w: no fields to update", domain.ErrInvalidInput)
	}

	sets = append(sets, "schema_version = schema_version + 1, updated_at = now()")

	query := fmt.Sprintf(`UPDATE entity_types SET %s WHERE name = $1`, strings.Join(sets, ", "))
	finalArgs := append([]any{req.Name}, args...)

	result, err := tx.Exec(ctx, query, finalArgs...)
	if err != nil {
		return nil, err
	}
	if result.RowsAffected() == 0 {
		return nil, fmt.Errorf("%w: entity type %q", domain.ErrNotFound, req.Name)
	}

	// Re-read.
	var et gen.EntityType
	var schemaBytes []byte
	err = tx.QueryRow(ctx, `
		SELECT name, parent_type, is_abstract, domain, layer, description,
		       lifecycle_id, attribute_schema, schema_version, status
		FROM entity_types WHERE name = $1`, req.Name).
		Scan(&et.Name, &et.ParentType, &et.IsAbstract, &et.Domain, &et.Layer,
			&et.Description, &et.LifecycleId, &schemaBytes, &et.SchemaVersion, &et.Status)
	if err != nil {
		return nil, err
	}
	var schema map[string]any
	if len(schemaBytes) > 0 && json.Unmarshal(schemaBytes, &schema) == nil && schema != nil {
		et.AttributeSchema = &schema
	}

	actorType, actor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, sqlcgen.New(tx), actorType, actor, "patch",
		nil, "PATCH", "/api/v1/ontology/entity-types/"+req.Name, "",
		map[string]any{"status": req.Body.Status}); auditErr != nil {
		return nil, auditErr
	}

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

	return gen.PatchEntityType200JSONResponse(et), nil
}

// ─── Metrics ───────────────────────────────────────────────────────────

func (s *Server) QueryMetrics(ctx context.Context, req gen.QueryMetricsRequestObject) (gen.QueryMetricsResponseObject, error) {
	if req.Params.EntityId == nil || *req.Params.EntityId == "" {
		return nil, fmt.Errorf("%w: entity_id is required", domain.ErrInvalidInput)
	}
	if req.Params.Metric == nil || len(*req.Params.Metric) == 0 {
		return nil, fmt.Errorf("%w: metric is required", domain.ErrInvalidInput)
	}

	entityID, err := s.resolveEntityID(ctx, *req.Params.EntityId)
	if err != nil {
		return nil, err
	}

	from := time.Now().Add(-24 * time.Hour)
	if req.Params.From != nil {
		from = *req.Params.From
	}
	to := time.Now()
	if req.Params.To != nil {
		to = *req.Params.To
	}

	items := []gen.MetricSeries{}
	for _, metricName := range *req.Params.Metric {
		series := gen.MetricSeries{
			EntityId: entityID.String(),
			Metric:   metricName,
			Rollup:   gen.MetricSeriesRollupRaw,
		}

		rows, err := s.pool.Query(ctx, `
			SELECT ts, value
			FROM metric_samples
			WHERE entity_id = $1 AND metric = $2
			  AND ts >= $3 AND ts <= $4
			ORDER BY ts ASC`,
			entityID, metricName, from, to)
		if err != nil {
			return nil, err
		}

		samples := []struct {
			Avg   *float32  `json:"avg"`
			Count *int      `json:"count"`
			Max   *float32  `json:"max"`
			Min   *float32  `json:"min"`
			Ts    time.Time `json:"ts"`
			Value *float32  `json:"value"`
		}{}

		for rows.Next() {
			var ts time.Time
			var val float64
			if err := rows.Scan(&ts, &val); err != nil {
				rows.Close()
				return nil, err
			}
			f := float32(val)
			samples = append(samples, struct {
				Avg   *float32  `json:"avg"`
				Count *int      `json:"count"`
				Max   *float32  `json:"max"`
				Min   *float32  `json:"min"`
				Ts    time.Time `json:"ts"`
				Value *float32  `json:"value"`
			}{Value: &f, Ts: ts})
		}
		rows.Close()
		if rows.Err() != nil {
			return nil, rows.Err()
		}

		series.Samples = samples
		items = append(items, series)
	}

	if items == nil {
		items = []gen.MetricSeries{}
	}
	return gen.QueryMetrics200JSONResponse{Items: items}, nil
}

func (s *Server) GetTrends(ctx context.Context, req gen.GetTrendsRequestObject) (gen.GetTrendsResponseObject, error) {
	entityID, err := s.resolveEntityID(ctx, req.EntityId)
	if err != nil {
		return nil, err
	}

	from := time.Now().Add(-7 * 24 * time.Hour)
	if req.Params.From != nil {
		from = *req.Params.From
	}

	rows, err := s.pool.Query(ctx, `
		SELECT metric,
		       ROUND(avg(value)::numeric, 2) AS avg_val,
		       ROUND(stddev(value)::numeric, 2) AS std_val,
		       count(*) AS sample_count,
		       ROUND(regr_slope(value, EXTRACT(EPOCH FROM ts)::numeric)::numeric, 4) AS slope
		FROM metric_samples
		WHERE entity_id = $1 AND ts >= $2
		GROUP BY metric
		ORDER BY metric`, entityID, from)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.Trend{}
	for rows.Next() {
		var t gen.Trend
		var avgVal, stdVal, slopeNum pgtype.Numeric
		var sampleCount int
		if err := rows.Scan(&t.Metric, &avgVal, &stdVal, &sampleCount, &slopeNum); err != nil {
			return nil, err
		}

		// Determine direction.
		if slopeNum.Valid {
			f, _ := slopeNum.Float64Value()
			t.Slope = float32Ptr(float32(f.Float64))
			if f.Float64 > 0.01 {
				t.Direction = gen.Improving
			} else if f.Float64 < -0.01 {
				t.Direction = gen.Degrading
			} else {
				t.Direction = gen.Stable
			}
		} else {
			t.Direction = gen.Unknown
		}
		items = append(items, t)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}
	if items == nil {
		items = []gen.Trend{}
	}
	return gen.GetTrends200JSONResponse{Items: items}, nil
}

func float32Ptr(f float32) *float32 {
	return &f
}

// ─── Agent Activity (stub) ─────────────────────────────────────────────

func (s *Server) QueryAgentActivity(ctx context.Context, request gen.QueryAgentActivityRequestObject) (gen.QueryAgentActivityResponseObject, error) {
	limit := clampLimit(request.Params.Limit)
	from := time.Now().Add(-24 * time.Hour)
	if request.Params.From != nil {
		from = *request.Params.From
	}
	to := time.Now()
	if request.Params.To != nil {
		to = *request.Params.To
	}

	var agentID *string
	if request.Params.AgentId != nil {
		a := *request.Params.AgentId
		agentID = &a
	}
	var activityType *string
	if request.Params.ActivityType != nil {
		a := string(*request.Params.ActivityType)
		activityType = &a
	}
	var entityID *string
	if request.Params.EntityId != nil {
		a := *request.Params.EntityId
		entityID = &a
	}
	var cursorID *int
	if request.Params.Cursor != nil && *request.Params.Cursor != "" {
		if id, err := parseIntOrZero(*request.Params.Cursor); err == nil && id > 0 {
			cursorID = &id
		}
	}

	rows, err := s.pool.Query(ctx, `
		SELECT id, ts, agent_id::text, session_id, activity_type, tool_name,
		       entity_id::text, input_summary, output_summary,
		       duration_ms, token_count, success, correlation_id
		FROM agent_activity
		WHERE ts >= $1 AND ts <= $2
		  AND ($3::text IS NULL OR agent_id::text = $3)
		  AND ($4::text IS NULL OR activity_type = $4)
		  AND ($5::text IS NULL OR entity_id::text = $5)
		  AND ($6::bigint IS NULL OR id < $6::bigint)
		ORDER BY id DESC
		LIMIT $7`,
		from, to, agentID, activityType, entityID, cursorID, limit+1)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	items := []gen.AgentActivity{}
	for rows.Next() {
		var a gen.AgentActivity
		if err := rows.Scan(&a.Id, &a.Ts, &a.AgentId, &a.SessionId,
			&a.ActivityType, &a.ToolName, &a.EntityId,
			&a.InputSummary, &a.OutputSummary,
			&a.DurationMs, &a.TokenCount, &a.Success,
			&a.CorrelationId); err != nil {
			return nil, err
		}
		items = append(items, a)
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}

	var next *string
	if len(items) > limit {
		items = items[:limit]
		lastID := fmt.Sprintf("%d", items[len(items)-1].Id)
		next = &lastID
	}
	if items == nil {
		items = []gen.AgentActivity{}
	}
	return gen.QueryAgentActivity200JSONResponse{Items: items, NextCursor: next}, nil
}

// ─── Helpers ───────────────────────────────────────────────────────────

func coalesceStr(s *string, def string) string {
	if s == nil || *s == "" {
		return def
	}
	return *s
}

func parseIntOrZero(s string) (int, error) {
	var n int
	for _, c := range s {
		if c < '0' || c > '9' {
			return 0, fmt.Errorf("invalid integer: %q", s)
		}
		n = n*10 + int(c-'0')
	}
	return n, nil
}