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 }