poimen-workflows/action/skills.go

package action

import (
	"context"
	"errors"
	"fmt"
	"os/exec"
	"strconv"
	"strings"
	"time"

	"go.temporal.io/sdk/activity"
	"github.com/rockliang/poimen/workflows/statemachine"
)

// PrepareSkillsInput is input to PrepareSkillsActivity.
type PrepareSkillsInput struct {
	Skills        []statemachine.SkillRef
	StreamTimeout time.Duration
	Provider      string // pi provider name (e.g. "homelab-reasoning"); required, pi has no usable default provider
}

// PrepareSkillsActivity prepares skills for use via pi command.
func PrepareSkillsActivity(ctx context.Context, in PrepareSkillsInput) error {
	if in.Provider == "" {
		return fmt.Errorf("PrepareSkillsInput.Provider must be set (pi has no usable default provider)")
	}

	for _, skill := range in.Skills {
		activity.RecordHeartbeat(ctx, skill.Name)

		// Run: pi clone-or-fetch <skill-url> --provider=<provider> --stream-timeout=<duration>
		cmd := exec.CommandContext(ctx, "pi", "clone-or-fetch", skill.URL, "--provider="+in.Provider, fmt.Sprintf("--stream-timeout=%s", in.StreamTimeout.String()))
		if err := cmd.Run(); err != nil {
			// Classify error
			classifiedErr := ClassifyPiErr(err, skill.Name)
			return classifiedErr
		}
	}

	return nil
}

// ClassifyPiErr classifies pi command errors into buckets.
// 4xx → NonRetryableApplicationError "PiClientError"
// 504 → ApplicationError "PiStreamTimeout" (retryable, but orchestrator learns and doubles timeout)
// 5xx (except 504) → generic retryable error
// Other → retryable
func ClassifyPiErr(err error, skillName string) error {
	if err == nil {
		return nil
	}

	// Try to extract HTTP status code from error message
	statusCode := ExtractHTTPStatus(err)
	if statusCode == 0 {
		// Not an HTTP error, return as-is for generic retry
		return fmt.Errorf("pi %s failed: %w", skillName, err)
	}

	switch {
	case statusCode >= 400 && statusCode < 500:
		// 4xx errors are non-retryable
		return NewNonRetryableApplicationError(
			fmt.Sprintf("PiClientError: status %d", statusCode),
			fmt.Sprintf("pi %s returned HTTP %d", skillName, statusCode),
		)
	case statusCode == 504:
		// 504 Gateway Timeout - stream timeout
		// This is retryable, but signals the orchestrator to double the timeout
		return NewApplicationError(
			fmt.Sprintf("PiStreamTimeout: status %d", statusCode),
			fmt.Sprintf("pi %s returned HTTP 504 (stream timeout)", skillName),
		)
	case statusCode >= 500:
		// Other 5xx errors are retryable
		return fmt.Errorf("pi %s returned HTTP %d: %w", skillName, statusCode, err)
	default:
		return fmt.Errorf("pi %s failed: %w", skillName, err)
	}
}

// ExtractHTTPStatus tries to extract HTTP status code from error message.
func ExtractHTTPStatus(err error) int {
	if err == nil {
		return 0
	}

	errStr := err.Error()

	// Try to find 3-digit numbers that could be HTTP status codes
	parts := strings.Fields(errStr)
	for i, part := range parts {
		// Check if part is a 3-digit number (HTTP status code)
		if len(part) >= 3 {
			code, err := strconv.Atoi(part[:3])
			if err == nil && code >= 100 && code < 600 {
				return code
			}
		}

		// Check if previous part is "status" or "HTTP"
		if i > 0 {
			prev := strings.ToLower(parts[i-1])
			if (prev == "status" || prev == "status:") && len(part) >= 3 {
				code, err := strconv.Atoi(part[:3])
				if err == nil && code >= 100 && code < 600 {
					return code
				}
			}
		}
	}

	return 0
}

// NewNonRetryableApplicationError creates a non-retryable application error.
func NewNonRetryableApplicationError(errType, errMsg string) error {
	return &NonRetryableApplicationError{
		errType: errType,
		errMsg:  errMsg,
	}
}

// NonRetryableApplicationError represents a non-retryable application error.
type NonRetryableApplicationError struct {
	errType string
	errMsg  string
}

func (e *NonRetryableApplicationError) Error() string {
	return fmt.Sprintf("%s: %s", e.errType, e.errMsg)
}

func (e *NonRetryableApplicationError) Type() string {
	return e.errType
}

// NewApplicationError creates a retryable application error with a specific type.
func NewApplicationError(errType, errMsg string) error {
	return &ApplicationError{
		errType: errType,
		errMsg:  errMsg,
	}
}

// ApplicationError represents a retryable application error with a type.
type ApplicationError struct {
	errType string
	errMsg  string
}

func (e *ApplicationError) Error() string {
	return fmt.Sprintf("%s: %s", e.errType, e.errMsg)
}

func (e *ApplicationError) Type() string {
	return e.errType
}

// IsPiStreamTimeout checks if an error is a PiStreamTimeout error.
func IsPiStreamTimeout(err error) bool {
	var appErr *ApplicationError
	if errors.As(err, &appErr) {
		return strings.Contains(appErr.errType, "PiStreamTimeout")
	}
	return false
}