hexis/internal/execution/engine.go

package execution

import (
	"context"
	"errors"
	"fmt"
	"regexp"
	"time"

	"github.com/kami/hexis/internal/domain"
	"github.com/kami/hexis/internal/nexusclient"
	"github.com/kami/hexis/internal/provider"
	"github.com/kami/hexis/internal/storage"
)

const defaultTimeout = 30 * time.Second

// targetLookupTimeout bounds the Nexus round-trip performed before an
// execution is created. It is short on purpose: the check sits in front of
// every execute, and a slow Nexus must surface as a fast 503, not a hang.
const targetLookupTimeout = 5 * time.Second

// Target validation failures. ECOSYSTEM-SPEC.md §4.3: "Hexis never accepts a
// free-text target. Ever." A target is acceptable only if it is a canonical
// `ent_` ID that Nexus confirms exists, is active, and whose type the
// capability declares it can act on.
var (
	ErrTargetMalformed    = errors.New("target_entity_id is not a canonical Nexus entity id")
	ErrTargetNotFound     = errors.New("target entity does not exist in Nexus")
	ErrTargetNotActive    = errors.New("target entity is retired, merged or deleted")
	ErrTargetTypeMismatch = errors.New("target entity type is not accepted by this capability")
	// ErrTargetUnverifiable means Nexus could not be reached. Hexis fails
	// CLOSED here: an unreachable identity service must not degrade into
	// accepting arbitrary strings, which is precisely the hole this check
	// exists to close.
	ErrTargetUnverifiable = errors.New("target entity could not be verified: nexus unreachable")
)

// entityIDPattern is the canonical Nexus entity ID shape. Anything else is
// free text by definition and is rejected before any network call.
var entityIDPattern = regexp.MustCompile(`^ent_[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$`)

// EntityLookup is the slice of Nexus that Hexis needs in order to refuse
// free-text targets.
type EntityLookup interface {
	GetEntity(ctx context.Context, id string) (*nexusclient.Entity, error)
}

type Engine struct {
	store    storage.Interface
	registry *provider.Registry
	entities EntityLookup
}

type Option func(*Engine)

// WithEntityLookup wires the Nexus entity check. Production always supplies
// it (see cmd/hexisd). When it is absent — only in tests and in tooling that
// never executes — targets are still shape-checked, but existence and type
// cannot be verified.
func WithEntityLookup(lookup EntityLookup) Option {
	return func(e *Engine) { e.entities = lookup }
}

func New(store storage.Interface, registry *provider.Registry, opts ...Option) *Engine {
	e := &Engine{store: store, registry: registry}
	for _, opt := range opts {
		opt(e)
	}
	return e
}

// validateTarget enforces spec §4.3's "no free-text target" rule.
//
// Ordering matters: the shape check is free and runs first, so a garbage
// string never reaches Nexus. Existence and type are then checked against
// Nexus itself — Hexis holds no entity table of its own, so this is the only
// authority available.
//
// An empty capability.TargetTypes is NOT "anything goes by accident": it
// means the capability declares no type constraint (most workspace-MCP tools
// are global — `docker.list_containers` acts on the host, not on a typed
// entity). The existence and active-state checks still apply, so the target
// is always a real, canonical, live entity; only the type filter is skipped.
func (e *Engine) validateTarget(capability *domain.Capability, targetEntityID string) error {
	if !entityIDPattern.MatchString(targetEntityID) {
		return ErrTargetMalformed
	}

	// A capability pinned to one entity accepts nothing else, regardless of
	// what Nexus says.
	if capability.TargetEntityID != "" && capability.TargetEntityID != targetEntityID {
		return domain.ErrCapabilityNotBound
	}

	if e.entities == nil {
		return nil
	}

	ctx, cancel := context.WithTimeout(context.Background(), targetLookupTimeout)
	defer cancel()

	entity, err := e.entities.GetEntity(ctx, targetEntityID)
	if err != nil {
		if errors.Is(err, nexusclient.ErrEntityNotFound) {
			return ErrTargetNotFound
		}
		return fmt.Errorf("%w: %v", ErrTargetUnverifiable, err)
	}
	if entity.State != "" && entity.State != "active" {
		return fmt.Errorf("%w (state %q)", ErrTargetNotActive, entity.State)
	}

	if len(capability.TargetTypes) == 0 {
		return nil
	}
	for _, t := range capability.TargetTypes {
		if t == entity.Type {
			return nil
		}
	}
	return fmt.Errorf("%w: entity is %q, capability accepts %v", ErrTargetTypeMismatch, entity.Type, capability.TargetTypes)
}

type ExecuteResult struct {
	Execution *domain.Execution `json:"execution"`
}

func (e *Engine) Execute(req *domain.ExecuteRequest) (*ExecuteResult, error) {
	if req.IdempotencyKey != "" {
		existing, err := e.store.GetExecutionByIdempotencyKey(req.IdempotencyKey)
		if err == nil {
			return &ExecuteResult{Execution: existing}, nil
		}
	}

	capability, err := e.store.GetCapability(req.CapabilityID)
	if err != nil {
		return nil, fmt.Errorf("capability: %w", err)
	}

	if !capability.Enabled {
		return nil, domain.ErrCapabilityDisabled
	}

	// Validate the target before anything with a side effect — including
	// before consuming a confirmation, so a bad target never burns one.
	if err := e.validateTarget(capability, req.TargetEntityID); err != nil {
		return nil, err
	}

	if capability.RequiresConfirmation {
		if err := e.consumeConfirmation(req, capability); err != nil {
			return nil, err
		}
	}

	// One in-flight execution per (capability_id, target_entity_id). This check
	// is advisory — it gives a clean error before doing any work — but the
	// authoritative guard is the partial unique index enforced at INSERT below,
	// which closes the check-then-insert window between the two.
	if _, err := e.store.GetInFlightExecution(req.CapabilityID, req.TargetEntityID); err == nil {
		return nil, domain.ErrExecutionInFlight
	}

	prov, err := e.registry.Get(capability.Provider)
	if err != nil {
		return nil, fmt.Errorf("provider: %w", err)
	}

	now := time.Now().UTC()
	exec := &domain.Execution{
		ID:                 domain.NewExecutionID(),
		CapabilityID:       req.CapabilityID,
		CapabilityVersion:  capability.Version,
		TargetEntityID:     req.TargetEntityID,
		EntityVersion:      req.EntityVersion,
		Arguments:          req.Arguments,
		RequestedBy:        req.RequestedBy,
		Origin:             req.Origin,
		IdempotencyKey:     req.IdempotencyKey,
		ConfirmationID:     req.ConfirmationID,
		Status:             domain.ExecutionStarted,
		CorrelationID:      req.CorrelationID,
		CausationID:        req.CausationID,
		ResolutionEvidence: req.ResolutionEvidence,
		CreatedAt:          now,
		UpdatedAt:          now,
	}
	if exec.Arguments == nil {
		exec.Arguments = map[string]any{}
	}
	if exec.RequestedBy == nil {
		exec.RequestedBy = map[string]string{}
	}
	if exec.Origin == nil {
		exec.Origin = map[string]string{}
	}

	if err := e.store.CreateExecution(exec); err != nil {
		return nil, err
	}

	e.emitEventCorrelated(domain.EventExecutionStarted, exec.ID, exec.CorrelationID, exec.CausationID, map[string]any{
		"capability_id":    req.CapabilityID,
		"target_entity_id": req.TargetEntityID,
		"provider":         capability.Provider,
		"correlation_id":   req.CorrelationID,
	})

	timeout := defaultTimeout
	if capability.TimeoutSeconds > 0 {
		timeout = time.Duration(capability.TimeoutSeconds) * time.Second
	}

	result, execErr, timedOut := e.runWithTimeout(prov, capability, req, timeout)
	exec.UpdatedAt = time.Now().UTC()

	switch {
	case timedOut:
		// Outcome is never "failed" on timeout, and never retried automatically
		// per ECOSYSTEM-SPEC.md §4.3 — the side effect may or may not have landed.
		exec.Status = domain.ExecutionUnknown
		exec.Error = "execution timed out after " + timeout.String()
		exec.Result = map[string]any{"error": exec.Error}
		e.emitEventCorrelated(domain.EventExecutionFailed, exec.ID, exec.CorrelationID, exec.CausationID, map[string]any{
			"capability_id": req.CapabilityID,
			"outcome":       "unknown",
			"reason":        "timeout",
		})
	case execErr != nil:
		exec.Status = domain.ExecutionFailed
		exec.Error = execErr.Error()
		exec.Result = map[string]any{"error": execErr.Error()}
		e.emitEventCorrelated(domain.EventExecutionFailed, exec.ID, exec.CorrelationID, exec.CausationID, map[string]any{
			"capability_id": req.CapabilityID,
			"error":         execErr.Error(),
		})
	default:
		exec.Status = domain.ExecutionSucceeded
		exec.Result = result
		e.emitEventCorrelated(domain.EventExecutionSucceeded, exec.ID, exec.CorrelationID, exec.CausationID, map[string]any{
			"capability_id": req.CapabilityID,
		})
	}

	if err := e.store.UpdateExecution(exec); err != nil {
		return nil, err
	}

	return &ExecuteResult{Execution: exec}, nil
}

// runWithTimeout executes the provider call on a goroutine and returns
// (nil, nil, true) if it doesn't finish within timeout. The goroutine is not
// cancelled — providers don't accept a context today — so a slow executor
// keeps running in the background; its eventual result is discarded, never
// retried, and the caller has already moved on with outcome=unknown.
func (e *Engine) runWithTimeout(prov provider.Provider, capability *domain.Capability, req *domain.ExecuteRequest, timeout time.Duration) (map[string]any, error, bool) {
	type outcome struct {
		result map[string]any
		err    error
	}
	ch := make(chan outcome, 1)
	go func() {
		r, err := prov.Execute(capability, req)
		ch <- outcome{result: r, err: err}
	}()

	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()

	select {
	case o := <-ch:
		return o.result, o.err, false
	case <-ctx.Done():
		return nil, nil, true
	}
}

func (e *Engine) consumeConfirmation(req *domain.ExecuteRequest, capability *domain.Capability) error {
	if req.ConfirmationID == "" {
		return domain.ErrConfirmationRequired
	}
	conf, err := e.store.GetConfirmation(req.ConfirmationID)
	if err != nil {
		return err
	}
	if conf.State != domain.ConfirmationPending {
		return domain.ErrConfirmationConsumed
	}
	if time.Now().UTC().After(conf.ExpiresAt) {
		e.store.UpdateConfirmationState(conf.ID, domain.ConfirmationExpired)
		return domain.ErrConfirmationExpired
	}
	if conf.CapabilityID != capability.ID || conf.CapabilityVersion != capability.Version {
		return domain.ErrConfirmationInvalid
	}
	if conf.TargetEntityID != req.TargetEntityID {
		return domain.ErrConfirmationInvalid
	}
	normalized, err := domain.NormalizeArgs(req.Arguments)
	if err != nil {
		return err
	}
	if domain.HashArgs(normalized) != conf.ArgsHash {
		return domain.ErrConfirmationInvalid
	}

	return e.store.UpdateConfirmationState(conf.ID, domain.ConfirmationConsumed)
}

// CreateConfirmation validates a would-be execute request against a
// capability and mints a time-bounded confirmation for it.
func (e *Engine) CreateConfirmation(capabilityID, targetEntityID, requester string, args map[string]any) (*domain.Confirmation, error) {
	capability, err := e.store.GetCapability(capabilityID)
	if err != nil {
		return nil, fmt.Errorf("capability: %w", err)
	}
	// A confirmation binds a target, so the target must be real here too —
	// otherwise a confirmation could be minted for a free-text target and the
	// execute-time check would be the only thing standing between it and a
	// side effect.
	if err := e.validateTarget(capability, targetEntityID); err != nil {
		return nil, err
	}

	normalized, err := domain.NormalizeArgs(args)
	if err != nil {
		return nil, err
	}

	now := time.Now().UTC()
	conf := &domain.Confirmation{
		ID:                domain.NewConfirmationID(),
		CapabilityID:      capability.ID,
		CapabilityVersion: capability.Version,
		TargetEntityID:    targetEntityID,
		ArgsNormalized:    normalized,
		ArgsHash:          domain.HashArgs(normalized),
		Requester:         requester,
		CreatedAt:         now,
		ExpiresAt:         now.Add(domain.ConfirmationTTL),
		State:             domain.ConfirmationPending,
	}
	if err := e.store.CreateConfirmation(conf); err != nil {
		return nil, err
	}
	return conf, nil
}

func (e *Engine) ValidateTarget(capabilityID, entityID string) error {
	cap, err := e.store.GetCapability(capabilityID)
	if err != nil {
		return err
	}
	return e.validateTarget(cap, entityID)
}

func (e *Engine) emitEventCorrelated(evtType domain.HexisEventType, entityID, correlationID, causationID string, payload map[string]any) {
	e.store.AppendEvent(&domain.Event{
		ID:            domain.NewEventID(),
		Type:          evtType,
		Timestamp:     time.Now().UTC(),
		CorrelationID: correlationID,
		CausationID:   causationID,
		Payload:       payload,
	})
}

type ResolveRequest struct {
	Query string   `json:"query"`
	Types []string `json:"types,omitempty"`
}

type ResolveResult struct {
	Status     string           `json:"status"`
	Candidates []map[string]any `json:"candidates,omitempty"`
	EntityID   string           `json:"entity_id,omitempty"`
}

var _ json.Marshaler = (*ResolveResult)(nil)

func (r *ResolveResult) MarshalJSON() ([]byte, error) {
	m := map[string]any{"status": r.Status}
	if r.Candidates != nil {
		m["candidates"] = r.Candidates
	}
	if r.EntityID != "" {
		m["entity_id"] = r.EntityID
	}
	return json.Marshal(m)
}