refactor: split phase3.go + extract MCP tool registry (R4)

internal/httpapi/phase3.go (2627 lines, 12+ resource domains) split into 15 per-resource files: - actuator.go: SSH execution machinery (initSSH, sshExec, resolveRunTarget, executeApprovedAction, jsonErr, gatewayPreflightPassed, resolveTemplate) - checks.go, classifications.go, executions.go, approvals.go, patterns.go, skills.go, approval_rules.go, autonomy.go, risk_classes.go, relationships.go, entity_types.go, metrics.go, agent_activity.go, helpers.go — one file per resource domain, each with its own imports. internal/mcp/server.go: newServer (708 lines, 33 inline tool registrations) refactored to a registry pattern: - internal/mcp/tools.go (new): toolReg struct + allTools() returning all 33 tool definitions. Handler logic moved verbatim — no changes to tool names, descriptions, schemas, or behavior. - server.go: newServer is now 9 lines (iterate registry, AddTool each). -699 lines. No function logic, names, or signatures changed. go vet, build, and all tests pass (httpapi, mcp, db, policy).
2026-07-17 22:41:40 +02:00
parent a2410cf9c2
commit fb39a48bef
23 changed files with 3539 additions and 3342 deletions
--- a/internal/httpapi/executions.go
+++ b/internal/httpapi/executions.go
@@ -0,0 +1,267 @@
+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
+}