Refactor agent runtime into reconciler-based architecture

Extract task processing logic from runtime into dedicated TaskReconciler
component following controller pattern. This separation improves code
organization and enables better task lifecycle management.

Changes:

Architecture:
• Extract TaskReconciler from Runtime to handle all task processing
• Move task state machine logic into dedicated component
• Introduce TaskPhase enum (AwaitInput, InvokeModel, ExecuteTools, Suspended)
• Create ModelProviderFactory for centralizing provider client creation
• Extract SyncMap into separate file for reusability

Task Reconciliation:
• Implement computeStatus to analyze message history and determine next action
• Split task processing into phase-specific reconcilers
• Handle model invocation cancellation gracefully with proper state updates
• Mark assistant messages as processed immediately when no tool calls present
• Add structured error handling for each reconciliation phase

Runtime Simplification:
• Remove 700+ lines of task processing logic from Runtime
• Eliminate CancelTask method in favor of reconciler-managed cancellation
• Update AgentRuntime interface to remove CancelTask dependency
• Replace direct queue management with TaskReconciler delegation
• Increase default concurrency from 5 to 50 workers

API Updates:
• Implement SuspendTask by updating task phase in database
• Remove runtime cancellation calls in favor of database-driven suspension
• Publish task events after phase transitions

This refactoring improves separation of concerns, makes task processing
more testable, and provides clearer boundaries between runtime management
and task execution logic.

Co-authored-by: construct-agent <noreply@construct.sh>

Author: Furisto

backend/agent/client.go

@@ -0,0 +1,81 @@
+package agent
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/furisto/construct/backend/memory"
+	"github.com/furisto/construct/backend/memory/schema/types"
+	"github.com/furisto/construct/backend/model"
+	"github.com/furisto/construct/backend/secret"
+	"github.com/google/uuid"
+)
+
+type ModelProviderFactory struct {
+	encryption *secret.Client
+	memory     *memory.Client
+}
+
+func NewModelProviderFactory(encryption *secret.Client, memory *memory.Client) *ModelProviderFactory {
+	return &ModelProviderFactory{
+		encryption: encryption,
+		memory:     memory,
+	}
+}
+
+func (f *ModelProviderFactory) CreateClient(
+	ctx context.Context,
+	modelProviderID uuid.UUID,
+) (model.ModelProvider, error) {
+	provider, err := f.memory.ModelProvider.Get(ctx, modelProviderID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch model provider: %w", err)
+	}
+
+	providerAuth, err := f.encryption.Decrypt(provider.Secret, []byte(secret.ModelProviderSecret(provider.ID)))
+	if err != nil {
+		return nil, fmt.Errorf("failed to decrypt model provider secret: %w", err)
+	}
+
+	var auth struct {
+		APIKey string `json:"apiKey"`
+	}
+	err = json.Unmarshal(providerAuth, &auth)
+	if err != nil {
+		return nil, fmt.Errorf("failed to unmarshal model provider auth: %w", err)
+	}
+
+	switch provider.ProviderType {
+	case types.ModelProviderTypeAnthropic:
+		apiProvider, err := model.NewAnthropicProvider(auth.APIKey)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create Anthropic provider: %w", err)
+		}
+		return apiProvider, nil
+
+	case types.ModelProviderTypeOpenAI:
+		apiProvider, err := model.NewOpenAICompletionProvider(auth.APIKey)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create OpenAI provider: %w", err)
+		}
+		return apiProvider, nil
+
+	case types.ModelProviderTypeGemini:
+		apiProvider, err := model.NewGeminiProvider(auth.APIKey)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create Gemini provider: %w", err)
+		}
+		return apiProvider, nil
+
+	case types.ModelProviderTypeXAI:
+		apiProvider, err := model.NewOpenAICompletionProvider(auth.APIKey, model.WithURL("https://api.xai.com/v1"))
+		if err != nil {
+			return nil, fmt.Errorf("failed to create XAI provider: %w", err)
+		}
+		return apiProvider, nil
+
+	default:
+		return nil, fmt.Errorf("unknown model provider type: %s", provider.ProviderType)
+	}
+}

backend/agent/conv.go

@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"log/slog"
-	"strings"
 
 	v1 "github.com/furisto/construct/api/go/v1"
 	"github.com/furisto/construct/backend/memory"
@@ -84,11 +83,8 @@ func ConvertMemoryMessageBlocksToModel(blocks []types.MessageBlock) ([]model.Con
 			}
 
 			result := interpreterResult.Output
-			if interpreterResult.Error != ""{
+			if interpreterResult.Error != "" {
 				result = interpreterResult.Output + "\n\n" + interpreterResult.Error
-				if strings.Contains(interpreterResult.Error, "ReferenceError") {
-					result = result + "\n" + "You likely tried to reference a variable from a previous code interpreter call. All code interpreter calls are isolated and share no state with each other."
-				}
 			}
 			contentBlocks = append(contentBlocks, &model.ToolResultBlock{
 				ID:        interpreterResult.ID,
@@ -136,7 +132,7 @@ func ConvertMemoryMessageToProto(m *memory.Message) (*v1.Message, error) {
 				return nil, fmt.Errorf("failed to unmarshal code interpreter call block: %w", err)
 			}
 
-			var interpreterArgs codeact.InterpreterArgs
+			var interpreterArgs codeact.InterpreterInput
 			err = json.Unmarshal(toolCall.Args, &interpreterArgs)
 			if err != nil {
 				return nil, fmt.Errorf("failed to unmarshal code interpreter args: %w", err)
@@ -696,3 +692,16 @@ func ConvertModelUsageToMemory(usage *model.Usage) *types.MessageUsage {
 		CacheWriteTokens: usage.CacheWriteTokens,
 	}
 }
+
+func convertTaskPhaseToMemory(phase TaskPhase) types.TaskPhase {
+	switch phase {
+	case TaskPhaseAwaitInput:
+		return types.TaskPhaseAwaiting
+	case TaskPhaseExecuteTools, TaskPhaseInvokeModel:
+		return types.TaskPhaseRunning
+	case TaskPhaseSuspended:
+		return types.TaskPhaseSuspended
+	}
+
+	return types.TaskPhaseUnspecified
+}

backend/agent/queue_metrics.go

@@ -16,7 +16,7 @@ type workqueueMetricsProvider struct {
 	retries                 *prometheus.CounterVec
 }
 
-func newWorkqueueMetricsProvider(registry *prometheus.Registry) *workqueueMetricsProvider {
+func newWorkqueueMetricsProvider(registry prometheus.Registerer) *workqueueMetricsProvider {
 	p := &workqueueMetricsProvider{
 		depth: prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Subsystem: "workqueue",