oikos/internal/httpapi/executions.go

package httpapi

import (
	"context"
	"encoding/json"
	"fmt"

	"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"
)

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

	// Full UUID, not a truncated prefix — an 8-char prefix of a UUIDv7
	// collides for real under back-to-back requests since the leading bytes
	// encode a millisecond timestamp (observed live via the MCP run tool).
	execSlug := "exec:" + id.String()
	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
}