(workflow) add simple harness workflow for manual testing

action/skills.go

@@ -1,3 +1,163 @@
 package action
 
-// Empty stub - will be filled in T0.4
+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
+}
+
+// PrepareSkillsActivity prepares skills for use via pi command.
+func PrepareSkillsActivity(ctx context.Context, in PrepareSkillsInput) error {
+	for _, skill := range in.Skills {
+		activity.RecordHeartbeat(ctx, skill.Name)
+
+		// Run: pi clone-or-fetch <skill-url> --stream-timeout=<duration>
+		cmd := exec.CommandContext(ctx, "pi", "clone-or-fetch", skill.URL, 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
+}