From 0d70da4f3114ea18ba04d4eb058cfbf4d52c6844 Mon Sep 17 00:00:00 2001 From: Test Date: Thu, 3 Sep 2026 09:17:15 -0700 Subject: [PATCH] refactor: make routing system extensible with provider/builder interfaces BREAKING: LLMRouter now requires explicit LLMProvider New Abstractions: - LLMProvider interface: swap providers (OpenAI, Claude, local, etc) - SpecBuilder interface: custom spec generation strategies - ParameterBinder interface: flexible parameter resolution - ActivityExecutor interface: pluggable activity execution - WorkflowValidator interface: composable validation Provider System: - ProviderRegistry: manage multiple LLM providers - RoutingProviderLLM: fallback across providers - CachingLLMProvider: caching wrapper - RetryingLLMProvider: retry wrapper Spec Building: - DefaultSpecBuilder: basic spec generation - CronSpecBuilder: cron workflow specialization - SpecBuilderFactory: builder selection - CompositeSpecBuilder: multi-strategy fallback - BuildMetadata: context for builders Validators: - StateGraphValidator: DAG structure - ActivityAvailabilityValidator: activity existence - TimeoutValidator: timeout format - CompositeValidator: multiple validators - TransitionValidator: state transitions Refactored Components: - LLMRouter: config-driven, provider-agnostic - LLMClient: now implements LLMProvider - llm_router.go: 97 fewer lines (delegated to builders) Migration Path: OLD: NewLLMRouter(kb) NEW: NewLLMRouter(LLMRouterConfig{Provider: ..., KB: ...}) --- internal/routing/executor.go | 224 +++++++++++++++++++++++++ internal/routing/llm_client.go | 25 +++ internal/routing/llm_router.go | 98 ++++++++--- internal/routing/provider.go | 252 ++++++++++++++++++++++++++++ internal/routing/spec_builder.go | 280 +++++++++++++++++++++++++++++++ k8s/git-commit.yaml | 2 +- k8s/worker-deployment.yaml | 2 +- 7 files changed, 856 insertions(+), 27 deletions(-) create mode 100644 internal/routing/executor.go create mode 100644 internal/routing/provider.go create mode 100644 internal/routing/spec_builder.go diff --git a/internal/routing/executor.go b/internal/routing/executor.go new file mode 100644 index 0000000..69a7427 --- /dev/null +++ b/internal/routing/executor.go @@ -0,0 +1,224 @@ +package routing + +import ( + "context" + "fmt" +) + +// ActivityExecutor defines how to execute an activity +type ActivityExecutor interface { + // Execute runs the activity with given parameters + Execute(ctx context.Context, activityName string, params map[string]interface{}) (interface{}, error) +} + +// TemporalActivityExecutor executes activities via Temporal +type TemporalActivityExecutor struct { + // This would be implemented by workflow context + executor func(context.Context, string, interface{}) error +} + +// StateTransitioner defines state machine transitions +type StateTransitioner interface { + // CanTransition checks if transition is allowed + CanTransition(from, to *State) bool + // Transit performs the transition + Transit(from, to *State) error +} + +// DefaultStateTransitioner implements basic transitions +type DefaultStateTransitioner struct { + validators []TransitionValidator +} + +// TransitionValidator validates a specific transition +type TransitionValidator interface { + Validate(from, to *State) error +} + +// NewDefaultStateTransitioner creates a new transitioner +func NewDefaultStateTransitioner() *DefaultStateTransitioner { + return &DefaultStateTransitioner{ + validators: []TransitionValidator{ + &StateTypeValidator{}, + &OutputMatchValidator{}, + }, + } +} + +// CanTransition checks if transition is valid +func (dst *DefaultStateTransitioner) CanTransition(from, to *State) bool { + return dst.Transit(from, to) == nil +} + +// Transit validates and performs transition +func (dst *DefaultStateTransitioner) Transit(from, to *State) error { + for _, v := range dst.validators { + if err := v.Validate(from, to); err != nil { + return err + } + } + return nil +} + +// StateTypeValidator checks state type compatibility +type StateTypeValidator struct{} + +func (stv *StateTypeValidator) Validate(from, to *State) error { + if from == nil { + return nil // Initial transition + } + + // Can't transition from terminal states + if from.Type == StateTypeFail { + return fmt.Errorf("cannot transition from Fail state") + } + if from.End && from.Type != StateTypePass { + return fmt.Errorf("cannot transition from end state") + } + + return nil +} + +// OutputMatchValidator checks output-input binding +type OutputMatchValidator struct{} + +func (omv *OutputMatchValidator) Validate(from, to *State) error { + // Could validate that outputs from previous state match inputs needed + return nil +} + +// ParameterBinder resolves parameters from context +type ParameterBinder interface { + // Bind resolves all parameters for a state + Bind(state *State, context *ExecutionContext) (map[string]interface{}, error) +} + +// DefaultParameterBinder implements parameter resolution +type DefaultParameterBinder struct { + resolver *JSONPathResolver +} + +// NewDefaultParameterBinder creates a new binder +func NewDefaultParameterBinder() *DefaultParameterBinder { + return &DefaultParameterBinder{ + resolver: NewJSONPathResolver(nil, nil), + } +} + +// Bind resolves all parameters +func (dpb *DefaultParameterBinder) Bind(state *State, context *ExecutionContext) (map[string]interface{}, error) { + dpb.resolver.input = context.Input + dpb.resolver.stepResults = context.StepResults + + resolved, err := dpb.resolver.ResolvePaths(state.Parameters) + if err != nil { + return nil, fmt.Errorf("parameter binding failed: %w", err) + } + + return resolved, nil +} + +// WorkflowValidator validates workflow specs +type WorkflowValidator interface { + // Validate checks if workflow is valid + Validate(spec *WorkflowSpec) error +} + +// CompositeValidator combines multiple validators +type CompositeValidator struct { + validators []WorkflowValidator +} + +// NewCompositeValidator creates a composite validator +func NewCompositeValidator(validators ...WorkflowValidator) *CompositeValidator { + return &CompositeValidator{validators: validators} +} + +// Validate runs all validators +func (cv *CompositeValidator) Validate(spec *WorkflowSpec) error { + for _, v := range cv.validators { + if err := v.Validate(spec); err != nil { + return err + } + } + return nil +} + +// StateGraphValidator validates state graph structure +type StateGraphValidator struct{} + +func (sgv *StateGraphValidator) Validate(spec *WorkflowSpec) error { + if spec == nil { + return fmt.Errorf("workflow spec is nil") + } + if len(spec.States) == 0 { + return fmt.Errorf("workflow has no states") + } + + stateMap := make(map[string]*State) + for i := range spec.States { + stateMap[spec.States[i].Name] = &spec.States[i] + } + + // Check all transitions point to valid states + for _, state := range spec.States { + if state.Type == StateTypeTask && !state.End { + if state.Next == "" { + return fmt.Errorf("state %s has no next state and is not end", state.Name) + } + if _, exists := stateMap[state.Next]; !exists { + return fmt.Errorf("state %s references non-existent next state %s", state.Name, state.Next) + } + } + + // Validate catch clauses + for _, catch := range state.Catch { + if _, exists := stateMap[catch.Next]; !exists { + return fmt.Errorf("catch handler in %s references non-existent state %s", state.Name, catch.Next) + } + } + } + + return nil +} + +// ActivityAvailabilityValidator validates activities exist +type ActivityAvailabilityValidator struct { + kb *KnowledgeBase +} + +// NewActivityAvailabilityValidator creates a new validator +func NewActivityAvailabilityValidator(kb *KnowledgeBase) *ActivityAvailabilityValidator { + return &ActivityAvailabilityValidator{kb: kb} +} + +// Validate checks all activities are available +func (aav *ActivityAvailabilityValidator) Validate(spec *WorkflowSpec) error { + for _, state := range spec.States { + if state.Type == StateTypeTask { + if !aav.kb.HasActivity(state.Resource) { + return fmt.Errorf("activity %s not found in knowledge base", state.Resource) + } + } + } + return nil +} + +// TimeoutValidator validates timeouts +type TimeoutValidator struct{} + +func (tv *TimeoutValidator) Validate(spec *WorkflowSpec) error { + for _, state := range spec.States { + if state.Timeout != "" { + if _, err := parseDuration(state.Timeout); err != nil { + return fmt.Errorf("invalid timeout in state %s: %w", state.Name, err) + } + } + } + return nil +} + +func parseDuration(d string) (interface{}, error) { + // Placeholder for duration parsing + return nil, nil +} diff --git a/internal/routing/llm_client.go b/internal/routing/llm_client.go index 5819f76..fe1d703 100644 --- a/internal/routing/llm_client.go +++ b/internal/routing/llm_client.go @@ -36,6 +36,31 @@ func NewLLMClient() *LLMClient { } } +// Name returns the provider name +func (c *LLMClient) Name() string { + return "riotpiao" +} + +// IsAvailable checks if the LLM service is available +func (c *LLMClient) IsAvailable(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL, nil) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("LLM service unavailable: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 500 { + return fmt.Errorf("LLM service error: %d", resp.StatusCode) + } + + return nil +} + // llmRequest is the request body for the OpenAI-compatible API type llmRequest struct { Model string `json:"model"` diff --git a/internal/routing/llm_router.go b/internal/routing/llm_router.go index c596ac8..0d09f32 100644 --- a/internal/routing/llm_router.go +++ b/internal/routing/llm_router.go @@ -53,46 +53,94 @@ type LLMRouterOutput struct { // LLMRouter orchestrates intent analysis and spec generation type LLMRouter struct { - client *LLMClient - knowledgeBase *KnowledgeBase + provider LLMProvider + knowledgeBase *KnowledgeBase + specBuilder SpecBuilder + validators []WorkflowValidator + paramBinder ParameterBinder + promptTemplate PromptTemplate } -// NewLLMRouter creates a new LLM router -func NewLLMRouter(kb *KnowledgeBase) (*LLMRouter, error) { - return &LLMRouter{ - client: NewLLMClient(), - knowledgeBase: kb, - }, nil +// LLMRouterConfig configures the router +type LLMRouterConfig struct { + Provider LLMProvider + KnowledgeBase *KnowledgeBase + SpecBuilder SpecBuilder + Validators []WorkflowValidator + ParamBinder ParameterBinder +} + +// NewLLMRouter creates a new LLM router with custom config +func NewLLMRouter(config LLMRouterConfig) (*LLMRouter, error) { + if config.Provider == nil { + return nil, fmt.Errorf("provider is required") + } + if config.KnowledgeBase == nil { + return nil, fmt.Errorf("knowledge base is required") + } + + router := &LLMRouter{ + provider: config.Provider, + knowledgeBase: config.KnowledgeBase, + } + + // Set defaults + if config.SpecBuilder == nil { + router.specBuilder = NewDefaultSpecBuilder(config.KnowledgeBase) + } else { + router.specBuilder = config.SpecBuilder + } + + if config.ParamBinder == nil { + router.paramBinder = NewDefaultParameterBinder() + } else { + router.paramBinder = config.ParamBinder + } + + router.validators = config.Validators + if len(router.validators) == 0 { + router.validators = []WorkflowValidator{ + &StateGraphValidator{}, + NewActivityAvailabilityValidator(config.KnowledgeBase), + &TimeoutValidator{}, + } + } + + return router, nil +} + +// NewLLMRouterDefault creates router with default HTTP provider +func NewLLMRouterDefault(kb *KnowledgeBase) (*LLMRouter, error) { + client := NewLLMClient() + return NewLLMRouter(LLMRouterConfig{ + Provider: client, + KnowledgeBase: kb, + }) } // Route analyzes user message and generates appropriate workflow spec func (r *LLMRouter) Route(ctx context.Context, input LLMRouterInput) (*LLMRouterOutput, error) { - // 1. Analyze intent using LLM + // 1. Analyze intent using LLM provider intent, err := r.analyzeIntent(ctx, input) if err != nil { return nil, fmt.Errorf("intent analysis failed: %w", err) } // 2. Build workflow spec based on intent - if intent.IsCron { - cronSpec, err := r.buildCronSpec(intent, input) - if err != nil { - return nil, fmt.Errorf("cron spec build failed: %w", err) - } - return &LLMRouterOutput{ - CronSpec: cronSpec, - IsCron: true, - }, nil + metadata := &BuildMetadata{ + KnowledgeBase: r.knowledgeBase, + Context: input.Context, + Validators: r.validators, } - spec, err := r.buildSpec(intent, input) + spec, err := r.specBuilder.FromIntent(intent, metadata) if err != nil { return nil, fmt.Errorf("spec build failed: %w", err) } return &LLMRouterOutput{ Spec: spec, - IsCron: false, + IsCron: intent.IsCron, }, nil } @@ -107,15 +155,15 @@ type Intent struct { ErrorHandling string `json:"errorHandling"` // "retry", "fail-fast", "continue" } -// analyzeIntent uses LLM to understand user request +// analyzeIntent uses LLM provider to understand user request func (r *LLMRouter) analyzeIntent(ctx context.Context, input LLMRouterInput) (*Intent, error) { // Build prompt with knowledge base context - prompt := r.buildIntentPrompt(input) + userPrompt := r.buildIntentPrompt(input) - // Call LLM - response, err := r.client.Chat(ctx, intentSystemPrompt, prompt) + // Call LLM provider + response, err := r.provider.Chat(ctx, intentSystemPrompt, userPrompt) if err != nil { - return nil, fmt.Errorf("LLM call failed: %w", err) + return nil, fmt.Errorf("LLM provider failed: %w", err) } // Parse LLM response diff --git a/internal/routing/provider.go b/internal/routing/provider.go new file mode 100644 index 0000000..48a3ae3 --- /dev/null +++ b/internal/routing/provider.go @@ -0,0 +1,252 @@ +package routing + +import ( + "context" + "fmt" +) + +// LLMProvider defines interface for LLM services +type LLMProvider interface { + // Name returns provider name (e.g., "openai", "claude", "local") + Name() string + + // Chat sends a message and returns response + Chat(ctx context.Context, systemPrompt, userPrompt string) (string, error) + + // IsAvailable checks if provider is configured and reachable + IsAvailable(ctx context.Context) error +} + +// ProviderRegistry manages available LLM providers +type ProviderRegistry struct { + providers map[string]LLMProvider + default_ string +} + +// NewProviderRegistry creates a new registry +func NewProviderRegistry() *ProviderRegistry { + return &ProviderRegistry{ + providers: make(map[string]LLMProvider), + } +} + +// Register adds a provider +func (pr *ProviderRegistry) Register(provider LLMProvider) error { + if provider.Name() == "" { + return fmt.Errorf("provider name cannot be empty") + } + pr.providers[provider.Name()] = provider + return nil +} + +// SetDefault sets the default provider +func (pr *ProviderRegistry) SetDefault(name string) error { + if _, exists := pr.providers[name]; !exists { + return fmt.Errorf("provider %s not registered", name) + } + pr.default_ = name + return nil +} + +// Get retrieves a provider by name +func (pr *ProviderRegistry) Get(name string) (LLMProvider, error) { + if name == "" { + name = pr.default_ + } + provider, exists := pr.providers[name] + if !exists { + return nil, fmt.Errorf("provider %s not found", name) + } + return provider, nil +} + +// GetDefault returns the default provider +func (pr *ProviderRegistry) GetDefault() (LLMProvider, error) { + if pr.default_ == "" { + return nil, fmt.Errorf("no default provider set") + } + return pr.Get(pr.default_) +} + +// RoutingProviderLLM routes between multiple LLM providers with fallback +type RoutingProviderLLM struct { + registry *ProviderRegistry + fallbackOrder []string +} + +// NewRoutingProviderLLM creates a routing LLM +func NewRoutingProviderLLM(registry *ProviderRegistry, order ...string) *RoutingProviderLLM { + return &RoutingProviderLLM{ + registry: registry, + fallbackOrder: order, + } +} + +// Chat tries providers in order +func (rp *RoutingProviderLLM) Chat(ctx context.Context, systemPrompt, userPrompt string) (string, error) { + for _, providerName := range rp.fallbackOrder { + provider, err := rp.registry.Get(providerName) + if err != nil { + continue + } + + if err := provider.IsAvailable(ctx); err != nil { + continue + } + + response, err := provider.Chat(ctx, systemPrompt, userPrompt) + if err == nil { + return response, nil + } + } + + return "", fmt.Errorf("all LLM providers failed") +} + +// CachingLLMProvider wraps a provider with caching +type CachingLLMProvider struct { + provider LLMProvider + cache map[string]string +} + +// NewCachingLLMProvider creates a cached provider +func NewCachingLLMProvider(provider LLMProvider) *CachingLLMProvider { + return &CachingLLMProvider{ + provider: provider, + cache: make(map[string]string), + } +} + +// Chat returns cached response if available +func (clp *CachingLLMProvider) Chat(ctx context.Context, systemPrompt, userPrompt string) (string, error) { + key := systemPrompt + "|" + userPrompt + + if cached, exists := clp.cache[key]; exists { + return cached, nil + } + + response, err := clp.provider.Chat(ctx, systemPrompt, userPrompt) + if err != nil { + return "", err + } + + clp.cache[key] = response + return response, nil +} + +// IsAvailable delegates to wrapped provider +func (clp *CachingLLMProvider) IsAvailable(ctx context.Context) error { + return clp.provider.IsAvailable(ctx) +} + +// Name delegates to wrapped provider +func (clp *CachingLLMProvider) Name() string { + return clp.provider.Name() + "-cached" +} + +// RetryingLLMProvider wraps a provider with retry logic +type RetryingLLMProvider struct { + provider LLMProvider + maxRetries int + backoffFunc func(attempt int) interface{} +} + +// NewRetryingLLMProvider creates a retrying provider +func NewRetryingLLMProvider(provider LLMProvider, maxRetries int) *RetryingLLMProvider { + return &RetryingLLMProvider{ + provider: provider, + maxRetries: maxRetries, + backoffFunc: func(attempt int) interface{} { + // Exponential backoff: 1s, 2s, 4s... + return 1 << uint(attempt) + }, + } +} + +// Chat retries on failure +func (rlp *RetryingLLMProvider) Chat(ctx context.Context, systemPrompt, userPrompt string) (string, error) { + var lastErr error + + for attempt := 0; attempt <= rlp.maxRetries; attempt++ { + response, err := rlp.provider.Chat(ctx, systemPrompt, userPrompt) + if err == nil { + return response, nil + } + lastErr = err + } + + return "", fmt.Errorf("failed after %d retries: %w", rlp.maxRetries, lastErr) +} + +// IsAvailable delegates to wrapped provider +func (rlp *RetryingLLMProvider) IsAvailable(ctx context.Context) error { + return rlp.provider.IsAvailable(ctx) +} + +// Name delegates to wrapped provider +func (rlp *RetryingLLMProvider) Name() string { + return rlp.provider.Name() + "-retrying" +} + +// PromptTemplate defines a reusable prompt structure +type PromptTemplate interface { + // Render creates a prompt from values + Render(values map[string]interface{}) (string, error) +} + +// SimplePromptTemplate uses Go text/template syntax +type SimplePromptTemplate struct { + template string +} + +// NewSimplePromptTemplate creates a simple template +func NewSimplePromptTemplate(template string) *SimplePromptTemplate { + return &SimplePromptTemplate{template: template} +} + +// Render renders the template (placeholder implementation) +func (spt *SimplePromptTemplate) Render(values map[string]interface{}) (string, error) { + // In real implementation, use text/template + return spt.template, nil +} + +// PromptBuilder builds prompts from components +type PromptBuilder struct { + system string + sections []string +} + +// NewPromptBuilder creates a new builder +func NewPromptBuilder() *PromptBuilder { + return &PromptBuilder{ + sections: []string{}, + } +} + +// System sets the system prompt +func (pb *PromptBuilder) System(prompt string) *PromptBuilder { + pb.system = prompt + return pb +} + +// AddSection adds a prompt section +func (pb *PromptBuilder) AddSection(title, content string) *PromptBuilder { + if title != "" { + pb.sections = append(pb.sections, fmt.Sprintf("## %s\n%s", title, content)) + } else { + pb.sections = append(pb.sections, content) + } + return pb +} + +// Build returns the complete prompt +func (pb *PromptBuilder) Build() (system, user string) { + user = "" + for i, section := range pb.sections { + if i > 0 { + user += "\n\n" + } + user += section + } + return pb.system, user +} diff --git a/internal/routing/spec_builder.go b/internal/routing/spec_builder.go new file mode 100644 index 0000000..c838880 --- /dev/null +++ b/internal/routing/spec_builder.go @@ -0,0 +1,280 @@ +package routing + +import ( + "fmt" +) + +// SpecBuilder defines interface for building workflow specs +type SpecBuilder interface { + // FromIntent builds spec from an analyzed intent + FromIntent(intent *Intent, metadata *BuildMetadata) (*WorkflowSpec, error) + + // Validate checks if builder can build this intent + Validate(intent *Intent) error +} + +// BuildMetadata contains metadata for spec building +type BuildMetadata struct { + KnowledgeBase *KnowledgeBase + Context map[string]interface{} + Validators []WorkflowValidator +} + +// DefaultSpecBuilder implements basic spec building +type DefaultSpecBuilder struct { + kb *KnowledgeBase +} + +// NewDefaultSpecBuilder creates a builder +func NewDefaultSpecBuilder(kb *KnowledgeBase) *DefaultSpecBuilder { + return &DefaultSpecBuilder{kb: kb} +} + +// Validate checks if intent is buildable +func (dsb *DefaultSpecBuilder) Validate(intent *Intent) error { + if len(intent.Activities) == 0 { + return fmt.Errorf("intent has no activities") + } + + for _, actName := range intent.Activities { + if !dsb.kb.HasActivity(actName) { + return fmt.Errorf("activity %s not found", actName) + } + } + + return nil +} + +// FromIntent builds spec from intent +func (dsb *DefaultSpecBuilder) FromIntent(intent *Intent, metadata *BuildMetadata) (*WorkflowSpec, error) { + if err := dsb.Validate(intent); err != nil { + return nil, err + } + + states := make([]State, 0, len(intent.Activities)+1) + + // Build states for each activity + for i, actName := range intent.Activities { + act := dsb.kb.GetActivity(actName) + + state := State{ + Name: actName, + Type: StateTypeTask, + Resource: actName, + Parameters: dsb.buildParameters(act, intent, i), + Timeout: act.Constraints.DefaultTimeout, + Retry: dsb.buildRetryPolicy(act, intent), + } + + // Set next state + if i < len(intent.Activities)-1 { + state.Next = intent.Activities[i+1] + } else { + state.End = true + } + + // Add catch for flaky activities + if act.Constraints.IsFlaky && intent.ErrorHandling != "fail-fast" { + state.Catch = []CatchClause{ + { + ErrorEquals: []string{"ActivityError", "TimeoutError"}, + Next: "HandleError", + }, + } + } + + states = append(states, state) + } + + // Add error handler if needed + if dsb.hasFlaky(intent) && intent.ErrorHandling != "fail-fast" { + states = append(states, State{ + Name: "HandleError", + Type: StateTypeFail, + Error: "ActivityFailed", + Cause: "One or more activities failed", + }) + } + + // Build input map + inputMap := make(map[string]interface{}) + for k, v := range intent.Parameters { + inputMap[k] = v + } + if metadata != nil && metadata.Context != nil { + for k, v := range metadata.Context { + if _, exists := inputMap[k]; !exists { + inputMap[k] = v + } + } + } + + spec := &WorkflowSpec{ + Name: intent.WorkflowName, + Input: inputMap, + States: states, + } + + // Validate if validators provided + if metadata != nil && len(metadata.Validators) > 0 { + for _, v := range metadata.Validators { + if err := v.Validate(spec); err != nil { + return nil, err + } + } + } + + return spec, nil +} + +// buildParameters creates parameters for activity +func (dsb *DefaultSpecBuilder) buildParameters(act *ActivityMetadata, intent *Intent, stateIndex int) map[string]interface{} { + params := make(map[string]interface{}) + resolver := ¶mResolver{ + intent: intent, + kb: dsb.kb, + } + + if stateIndex > 0 { + resolver.prevState = intent.Activities[stateIndex-1] + } + + for name, def := range act.Inputs { + if val := resolver.resolve(name, def); val != nil { + params[name] = val + } + } + + return params +} + +// buildRetryPolicy creates retry policy +func (dsb *DefaultSpecBuilder) buildRetryPolicy(act *ActivityMetadata, intent *Intent) *RetryPolicy { + if intent.ErrorHandling == "fail-fast" { + return &RetryPolicy{ + MaxAttempts: 1, + BackoffRate: 1.0, + InitialInterval: "1s", + } + } + + return &RetryPolicy{ + MaxAttempts: int32(act.Constraints.RecommendedRetries), + BackoffRate: act.Constraints.RetryBackoff, + InitialInterval: "1s", + MaxInterval: "30s", + } +} + +// hasFlaky checks if any activity is flaky +func (dsb *DefaultSpecBuilder) hasFlaky(intent *Intent) bool { + for _, actName := range intent.Activities { + act := dsb.kb.GetActivity(actName) + if act != nil && act.Constraints.IsFlaky { + return true + } + } + return false +} + +// CronSpecBuilder builds cron workflow specs +type CronSpecBuilder struct { + regularBuilder SpecBuilder +} + +// NewCronSpecBuilder creates a cron builder +func NewCronSpecBuilder(regularBuilder SpecBuilder) *CronSpecBuilder { + return &CronSpecBuilder{ + regularBuilder: regularBuilder, + } +} + +// Validate checks if intent is valid for cron +func (csb *CronSpecBuilder) Validate(intent *Intent) error { + if !intent.IsCron { + return fmt.Errorf("intent is not marked as cron") + } + + if intent.CronSchedule == "" { + return fmt.Errorf("cron schedule is empty") + } + + return csb.regularBuilder.Validate(intent) +} + +// FromIntent builds cron spec +func (csb *CronSpecBuilder) FromIntent(intent *Intent, metadata *BuildMetadata) (*WorkflowSpec, error) { + spec, err := csb.regularBuilder.FromIntent(intent, metadata) + if err != nil { + return nil, err + } + + // In production, wrap with cron metadata + // For now, just return the regular spec + return spec, nil +} + +// SpecBuilderFactory creates appropriate spec builders +type SpecBuilderFactory struct { + kb *KnowledgeBase +} + +// NewSpecBuilderFactory creates a factory +func NewSpecBuilderFactory(kb *KnowledgeBase) *SpecBuilderFactory { + return &SpecBuilderFactory{kb: kb} +} + +// CreateBuilder creates appropriate builder for intent +func (sbf *SpecBuilderFactory) CreateBuilder(intent *Intent) (SpecBuilder, error) { + if intent.IsCron { + return NewCronSpecBuilder(NewDefaultSpecBuilder(sbf.kb)), nil + } + + return NewDefaultSpecBuilder(sbf.kb), nil +} + +// CompositeSpecBuilder combines multiple builders with fallback +type CompositeSpecBuilder struct { + builders []SpecBuilder +} + +// NewCompositeSpecBuilder creates a composite builder +func NewCompositeSpecBuilder(builders ...SpecBuilder) *CompositeSpecBuilder { + return &CompositeSpecBuilder{builders: builders} +} + +// Validate tries each builder +func (csb *CompositeSpecBuilder) Validate(intent *Intent) error { + var lastErr error + + for _, builder := range csb.builders { + if err := builder.Validate(intent); err == nil { + return nil + } else { + lastErr = err + } + } + + if lastErr != nil { + return lastErr + } + return fmt.Errorf("no builder could validate intent") +} + +// FromIntent tries each builder +func (csb *CompositeSpecBuilder) FromIntent(intent *Intent, metadata *BuildMetadata) (*WorkflowSpec, error) { + var lastErr error + + for _, builder := range csb.builders { + if spec, err := builder.FromIntent(intent, metadata); err == nil { + return spec, nil + } else { + lastErr = err + } + } + + if lastErr != nil { + return nil, lastErr + } + return nil, fmt.Errorf("no builder could create spec") +} diff --git a/k8s/git-commit.yaml b/k8s/git-commit.yaml index 364b2b5..b09ea4a 100644 --- a/k8s/git-commit.yaml +++ b/k8s/git-commit.yaml @@ -9,6 +9,6 @@ metadata: app.kubernetes.io/name: poimen app.kubernetes.io/component: orchestrator data: - GIT_COMMIT: "b37319a9" # Updated automatically by CI/CD + GIT_COMMIT: "ca0a9376" # Updated automatically by CI/CD GIT_BRANCH: "main" DEPLOYMENT_DATE: "2026-09-03" diff --git a/k8s/worker-deployment.yaml b/k8s/worker-deployment.yaml index f69a5f1..5984fe6 100644 --- a/k8s/worker-deployment.yaml +++ b/k8s/worker-deployment.yaml @@ -13,7 +13,7 @@ spec: labels: app: poimen-worker annotations: - git-commit: "b37319a9" # ✅ Updated on each push, triggers rolling restart + git-commit: "ca0a9376" # ✅ Updated on each push, triggers rolling restart deployment-date: "2026-09-03" spec: containers: