diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index 2b64ce1526..76a59f3181 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -16,15 +16,19 @@ import ( "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/task/models" - "github.com/kandev/kandev/internal/task/repository" taskservice "github.com/kandev/kandev/internal/task/service" "github.com/kandev/kandev/internal/worktree" "github.com/kandev/kandev/pkg/api/v1" ) +// taskGetterRepo is the minimal interface needed by the scheduler adapter. +type taskGetterRepo interface { + GetTask(ctx context.Context, id string) (*models.Task, error) +} + // taskRepositoryAdapter adapts the task repository for the orchestrator's scheduler type taskRepositoryAdapter struct { - repo repository.Repository + repo taskGetterRepo svc *taskservice.Service } diff --git a/apps/backend/cmd/kandev/gateway.go b/apps/backend/cmd/kandev/gateway.go index 2cb4f3aeff..4612f563b9 100644 --- a/apps/backend/cmd/kandev/gateway.go +++ b/apps/backend/cmd/kandev/gateway.go @@ -20,7 +20,7 @@ import ( notificationstore "github.com/kandev/kandev/internal/notifications/store" "github.com/kandev/kandev/internal/orchestrator" orchestratorhandlers "github.com/kandev/kandev/internal/orchestrator/handlers" - "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" userservice "github.com/kandev/kandev/internal/user/service" ) @@ -52,7 +52,7 @@ func provideGateway( lifecycleMgr *lifecycle.Manager, agentRegistry *registry.Registry, notificationRepo notificationstore.Repository, - taskRepo repository.Repository, + taskRepo *sqliterepo.Repository, ) (*gateways.Gateway, *notificationservice.Service, *notificationcontroller.Controller, error) { gateway, err := gateways.Provide(log) if err != nil { diff --git a/apps/backend/cmd/kandev/helpers.go b/apps/backend/cmd/kandev/helpers.go index 7630dc3344..6a82253119 100644 --- a/apps/backend/cmd/kandev/helpers.go +++ b/apps/backend/cmd/kandev/helpers.go @@ -34,7 +34,7 @@ import ( spriteshandlers "github.com/kandev/kandev/internal/sprites" taskhandlers "github.com/kandev/kandev/internal/task/handlers" "github.com/kandev/kandev/internal/task/models" - "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" usercontroller "github.com/kandev/kandev/internal/user/controller" userhandlers "github.com/kandev/kandev/internal/user/handlers" @@ -62,7 +62,7 @@ func initDockerClient(ctx context.Context, cfg *config.Config, log *logger.Logge // buildSessionDataProvider constructs the session data provider function used by the WebSocket hub // to send initial data (git status, context window, available commands) when a client subscribes. -func buildSessionDataProvider(taskRepo repository.Repository, lifecycleMgr *lifecycle.Manager, log *logger.Logger) func(context.Context, string) ([]*ws.Message, error) { +func buildSessionDataProvider(taskRepo *sqliterepo.Repository, lifecycleMgr *lifecycle.Manager, log *logger.Logger) func(context.Context, string) ([]*ws.Message, error) { return func(ctx context.Context, sessionID string) ([]*ws.Message, error) { session, err := taskRepo.GetTaskSession(ctx, sessionID) if err != nil { @@ -78,7 +78,7 @@ func buildSessionDataProvider(taskRepo repository.Repository, lifecycleMgr *life } // appendGitStatusMessage adds a git status notification from the latest snapshot to result. -func appendGitStatusMessage(ctx context.Context, taskRepo repository.Repository, sessionID string, session *models.TaskSession, result []*ws.Message) []*ws.Message { +func appendGitStatusMessage(ctx context.Context, taskRepo *sqliterepo.Repository, sessionID string, session *models.TaskSession, result []*ws.Message) []*ws.Message { latestSnapshot, err := taskRepo.GetLatestGitSnapshot(ctx, sessionID) if err != nil || latestSnapshot == nil { return result @@ -252,7 +252,7 @@ type routeParams struct { gateway *gateways.Gateway dockerClient *docker.Client taskSvc *taskservice.Service - taskRepo repository.Repository + taskRepo *sqliterepo.Repository analyticsRepo analyticsrepository.Repository orchestratorSvc *orchestrator.Service lifecycleMgr *lifecycle.Manager diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index 11c4d05f34..6ef82069b1 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -10,7 +10,7 @@ import ( "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/secrets" - "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" userservice "github.com/kandev/kandev/internal/user/service" wfmodels "github.com/kandev/kandev/internal/workflow/models" @@ -21,7 +21,7 @@ import ( func provideOrchestrator( log *logger.Logger, eventBus bus.EventBus, - taskRepo repository.Repository, + taskRepo *sqliterepo.Repository, taskSvc *taskservice.Service, userSvc *userservice.Service, lifecycleMgr *lifecycle.Manager, diff --git a/apps/backend/cmd/kandev/services.go b/apps/backend/cmd/kandev/services.go index 48e8289c02..496c69aad0 100644 --- a/apps/backend/cmd/kandev/services.go +++ b/apps/backend/cmd/kandev/services.go @@ -35,7 +35,20 @@ func provideServices(cfg *config.Config, log *logger.Logger, repos *Repositories promptSvc := promptservice.NewService(repos.Prompts) workflowSvc := workflowservice.NewService(repos.Workflow, log) taskSvc := taskservice.NewService( - repos.Task, + taskservice.Repos{ + Workspaces: repos.Task, + Tasks: repos.Task, + TaskRepos: repos.Task, + Workflows: repos.Task, + Messages: repos.Task, + Turns: repos.Task, + Sessions: repos.Task, + GitSnapshots: repos.Task, + RepoEntities: repos.Task, + Executors: repos.Task, + Environments: repos.Task, + Reviews: repos.Task, + }, eventBus, log, taskservice.RepositoryDiscoveryConfig{ diff --git a/apps/backend/cmd/kandev/types.go b/apps/backend/cmd/kandev/types.go index b1bac5aee7..b3cb77bb3f 100644 --- a/apps/backend/cmd/kandev/types.go +++ b/apps/backend/cmd/kandev/types.go @@ -10,7 +10,7 @@ import ( promptservice "github.com/kandev/kandev/internal/prompts/service" promptstore "github.com/kandev/kandev/internal/prompts/store" "github.com/kandev/kandev/internal/secrets" - "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" userservice "github.com/kandev/kandev/internal/user/service" userstore "github.com/kandev/kandev/internal/user/store" @@ -19,7 +19,7 @@ import ( ) type Repositories struct { - Task repository.Repository + Task *sqliterepo.Repository Analytics analyticsrepository.Repository AgentSettings settingsstore.Repository User userstore.Repository diff --git a/apps/backend/internal/clarification/handlers.go b/apps/backend/internal/clarification/handlers.go index dd10240645..cc68ef7894 100644 --- a/apps/backend/internal/clarification/handlers.go +++ b/apps/backend/internal/clarification/handlers.go @@ -9,11 +9,18 @@ import ( "github.com/gin-gonic/gin" "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/task/repository" + taskmodels "github.com/kandev/kandev/internal/task/models" wsmsg "github.com/kandev/kandev/pkg/websocket" "go.uber.org/zap" ) +// messageStore is the minimal task repository interface required by clarification handlers. +type messageStore interface { + GetTaskSession(ctx context.Context, id string) (*taskmodels.TaskSession, error) + FindMessageByPendingID(ctx context.Context, pendingID string) (*taskmodels.Message, error) + UpdateMessage(ctx context.Context, message *taskmodels.Message) error +} + // Broadcaster interface for sending WebSocket notifications type Broadcaster interface { BroadcastToSession(sessionID string, msg *wsmsg.Message) @@ -32,12 +39,12 @@ type Handlers struct { store *Store hub Broadcaster messageCreator MessageCreator - repo repository.Repository + repo messageStore logger *logger.Logger } // NewHandlers creates new clarification handlers. -func NewHandlers(store *Store, hub Broadcaster, messageCreator MessageCreator, repo repository.Repository, log *logger.Logger) *Handlers { +func NewHandlers(store *Store, hub Broadcaster, messageCreator MessageCreator, repo messageStore, log *logger.Logger) *Handlers { return &Handlers{ store: store, hub: hub, @@ -48,7 +55,7 @@ func NewHandlers(store *Store, hub Broadcaster, messageCreator MessageCreator, r } // RegisterRoutes registers clarification HTTP routes. -func RegisterRoutes(router *gin.Engine, store *Store, hub Broadcaster, messageCreator MessageCreator, repo repository.Repository, log *logger.Logger) { +func RegisterRoutes(router *gin.Engine, store *Store, hub Broadcaster, messageCreator MessageCreator, repo messageStore, log *logger.Logger) { h := NewHandlers(store, hub, messageCreator, repo, log) api := router.Group("/api/v1/clarification") api.POST("/request", h.httpCreateRequest) diff --git a/apps/backend/internal/editors/service/service.go b/apps/backend/internal/editors/service/service.go index 1de4028e02..f111d4dbb9 100644 --- a/apps/backend/internal/editors/service/service.go +++ b/apps/backend/internal/editors/service/service.go @@ -17,10 +17,15 @@ import ( "github.com/kandev/kandev/internal/editors/models" "github.com/kandev/kandev/internal/editors/store" taskmodels "github.com/kandev/kandev/internal/task/models" - "github.com/kandev/kandev/internal/task/repository" usermodels "github.com/kandev/kandev/internal/user/models" ) +// taskSessionReader is the minimal task repository interface required by the editors service. +type taskSessionReader interface { + GetTaskSession(ctx context.Context, id string) (*taskmodels.TaskSession, error) + GetRepository(ctx context.Context, id string) (*taskmodels.Repository, error) +} + var ( ErrEditorNotFound = errors.New("editor not found") ErrEditorUnavailable = errors.New("editor not available") @@ -43,11 +48,11 @@ type UserSettingsProvider interface { type Service struct { repo store.Repository - taskRepo repository.Repository + taskRepo taskSessionReader userSettings UserSettingsProvider } -func NewService(repo store.Repository, taskRepo repository.Repository, userSettings UserSettingsProvider) *Service { +func NewService(repo store.Repository, taskRepo taskSessionReader, userSettings UserSettingsProvider) *Service { return &Service{ repo: repo, taskRepo: taskRepo, diff --git a/apps/backend/internal/integration/integration_test.go b/apps/backend/internal/integration/integration_test.go index 56ffc75e5d..9a8091a8f2 100644 --- a/apps/backend/internal/integration/integration_test.go +++ b/apps/backend/internal/integration/integration_test.go @@ -26,6 +26,7 @@ import ( gateways "github.com/kandev/kandev/internal/gateway/websocket" taskhandlers "github.com/kandev/kandev/internal/task/handlers" "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" "github.com/kandev/kandev/internal/workflow" workflowcontroller "github.com/kandev/kandev/internal/workflow/controller" @@ -38,7 +39,7 @@ import ( type TestServer struct { Server *httptest.Server Gateway *gateways.Gateway - TaskRepo repository.Repository + TaskRepo *sqliterepo.Repository TaskSvc *taskservice.Service EventBus bus.EventBus Logger *logger.Logger @@ -88,7 +89,12 @@ func NewTestServer(t *testing.T) *TestServer { require.NoError(t, err) // Initialize task service and wire workflow step creator - taskSvc := taskservice.NewService(taskRepo, eventBus, log, taskservice.RepositoryDiscoveryConfig{}) + taskSvc := taskservice.NewService(taskservice.Repos{ + Workspaces: taskRepo, Tasks: taskRepo, TaskRepos: taskRepo, + Workflows: taskRepo, Messages: taskRepo, Turns: taskRepo, + Sessions: taskRepo, GitSnapshots: taskRepo, RepoEntities: taskRepo, + Executors: taskRepo, Environments: taskRepo, Reviews: taskRepo, + }, eventBus, log, taskservice.RepositoryDiscoveryConfig{}) taskSvc.SetWorkflowStepCreator(workflowSvc) // Create WebSocket gateway diff --git a/apps/backend/internal/integration/test_server_test.go b/apps/backend/internal/integration/test_server_test.go index 8b71875135..418387d777 100644 --- a/apps/backend/internal/integration/test_server_test.go +++ b/apps/backend/internal/integration/test_server_test.go @@ -23,6 +23,7 @@ import ( taskhandlers "github.com/kandev/kandev/internal/task/handlers" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" "github.com/kandev/kandev/internal/workflow" workflowcontroller "github.com/kandev/kandev/internal/workflow/controller" @@ -36,7 +37,7 @@ import ( type OrchestratorTestServer struct { Server *httptest.Server Gateway *gateways.Gateway - TaskRepo repository.Repository + TaskRepo *sqliterepo.Repository TaskSvc *taskservice.Service WorkflowSvc *workflowservice.Service EventBus bus.EventBus @@ -49,7 +50,7 @@ type OrchestratorTestServer struct { // taskRepositoryAdapter adapts the task repository for the orchestrator type taskRepositoryAdapter struct { - repo repository.Repository + repo *sqliterepo.Repository svc *taskservice.Service } @@ -258,7 +259,12 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { require.NoError(t, err) // Initialize task service and wire workflow step creator - taskSvc := taskservice.NewService(taskRepo, eventBus, log, taskservice.RepositoryDiscoveryConfig{}) + taskSvc := taskservice.NewService(taskservice.Repos{ + Workspaces: taskRepo, Tasks: taskRepo, TaskRepos: taskRepo, + Workflows: taskRepo, Messages: taskRepo, Turns: taskRepo, + Sessions: taskRepo, GitSnapshots: taskRepo, RepoEntities: taskRepo, + Executors: taskRepo, Environments: taskRepo, Reviews: taskRepo, + }, eventBus, log, taskservice.RepositoryDiscoveryConfig{}) taskSvc.SetWorkflowStepCreator(workflowSvc) // Create simulated agent manager diff --git a/apps/backend/internal/notifications/service/service.go b/apps/backend/internal/notifications/service/service.go index a7c924a765..0e4e20c53e 100644 --- a/apps/backend/internal/notifications/service/service.go +++ b/apps/backend/internal/notifications/service/service.go @@ -13,7 +13,7 @@ import ( "github.com/kandev/kandev/internal/notifications/models" "github.com/kandev/kandev/internal/notifications/providers" notificationstore "github.com/kandev/kandev/internal/notifications/store" - "github.com/kandev/kandev/internal/task/repository" + taskmodels "github.com/kandev/kandev/internal/task/models" userstore "github.com/kandev/kandev/internal/user/store" "go.uber.org/zap" ) @@ -24,15 +24,20 @@ const ( var ErrProviderNotFound = errors.New("notification provider not found") +// taskGetter is the minimal repository interface needed by the notification service. +type taskGetter interface { + GetTask(ctx context.Context, id string) (*taskmodels.Task, error) +} + type Service struct { repo notificationstore.Repository - taskRepo repository.Repository + taskRepo taskGetter hub *gatewayws.Hub logger *logger.Logger providers map[models.ProviderType]providers.Provider } -func NewService(repo notificationstore.Repository, taskRepo repository.Repository, hub *gatewayws.Hub, log *logger.Logger) *Service { +func NewService(repo notificationstore.Repository, taskRepo taskGetter, hub *gatewayws.Hub, log *logger.Logger) *Service { providerMap := map[models.ProviderType]providers.Provider{ models.ProviderTypeLocal: providers.NewLocalProvider(hub), models.ProviderTypeApprise: providers.NewAppriseProvider(), diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index ca87dbf357..3487c727f6 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -14,6 +14,7 @@ import ( "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" wfmodels "github.com/kandev/kandev/internal/workflow/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -126,7 +127,7 @@ func testLogger() *logger.Logger { func strPtr(s string) *string { return &s } // setupTestRepo creates a real in-memory SQLite repository for testing. -func setupTestRepo(t *testing.T) repository.Repository { +func setupTestRepo(t *testing.T) *sqliterepo.Repository { t.Helper() tmpDir := t.TempDir() dbConn, err := db.OpenSQLite(filepath.Join(tmpDir, "test.db")) @@ -146,7 +147,7 @@ func setupTestRepo(t *testing.T) repository.Repository { } // seedSession creates a task, workspace, workflow and session in the repo for testing. -func seedSession(t *testing.T, repo repository.Repository, taskID, sessionID, workflowStepID string) { +func seedSession(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID, workflowStepID string) { t.Helper() ctx := context.Background() now := time.Now().UTC() @@ -194,11 +195,11 @@ func seedSession(t *testing.T, repo repository.Repository, taskID, sessionID, wo } // createTestService creates a Service with minimal dependencies for event handler testing. -func createTestService(repo repository.Repository, stepGetter *mockStepGetter, taskRepo *mockTaskRepo) *Service { +func createTestService(repo *sqliterepo.Repository, stepGetter *mockStepGetter, taskRepo *mockTaskRepo) *Service { return createTestServiceWithAgent(repo, stepGetter, taskRepo, &mockAgentManager{}) } -func createTestServiceWithAgent(repo repository.Repository, stepGetter *mockStepGetter, taskRepo *mockTaskRepo, agentMgr executor.AgentManagerClient) *Service { +func createTestServiceWithAgent(repo *sqliterepo.Repository, stepGetter *mockStepGetter, taskRepo *mockTaskRepo, agentMgr executor.AgentManagerClient) *Service { return &Service{ logger: testLogger(), repo: repo, diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 0f42fec217..4d555a51ba 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -10,11 +10,38 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/secrets" "github.com/kandev/kandev/internal/task/models" - "github.com/kandev/kandev/internal/task/repository" v1 "github.com/kandev/kandev/pkg/api/v1" "go.uber.org/zap" ) +// executorStore is the minimal repository interface required by the Executor. +type executorStore interface { + // Task + GetTask(ctx context.Context, id string) (*models.Task, error) + UpdateTaskState(ctx context.Context, id string, state v1.TaskState) error + // Task↔repo junction + GetPrimaryTaskRepository(ctx context.Context, taskID string) (*models.TaskRepository, error) + // Session + CreateTaskSession(ctx context.Context, session *models.TaskSession) error + GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) + UpdateTaskSession(ctx context.Context, session *models.TaskSession) error + UpdateTaskSessionState(ctx context.Context, id string, state models.TaskSessionState, errorMessage string) error + SetSessionPrimary(ctx context.Context, sessionID string) error + ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) + ListActiveTaskSessionsByTaskID(ctx context.Context, taskID string) ([]*models.TaskSession, error) + // Session worktree + CreateTaskSessionWorktree(ctx context.Context, sessionWorktree *models.TaskSessionWorktree) error + // Repository entity + GetRepository(ctx context.Context, id string) (*models.Repository, error) + // Executor + GetExecutor(ctx context.Context, id string) (*models.Executor, error) + GetExecutorProfile(ctx context.Context, id string) (*models.ExecutorProfile, error) + GetExecutorRunningBySessionID(ctx context.Context, sessionID string) (*models.ExecutorRunning, error) + UpsertExecutorRunning(ctx context.Context, running *models.ExecutorRunning) error + // Workspace + GetWorkspace(ctx context.Context, id string) (*models.Workspace, error) +} + // Common errors const defaultBaseBranch = "main" @@ -167,7 +194,7 @@ type TaskStateChangeFunc func(ctx context.Context, taskID string, state v1.TaskS // Executor manages agent execution for tasks type Executor struct { agentManager AgentManagerClient - repo repository.Repository + repo executorStore secretStore secrets.SecretStore shellPrefs ShellPreferenceProvider logger *logger.Logger @@ -197,7 +224,7 @@ type ShellPreferenceProvider interface { } // NewExecutor creates a new executor -func NewExecutor(agentManager AgentManagerClient, repo repository.Repository, log *logger.Logger, cfg ExecutorConfig) *Executor { +func NewExecutor(agentManager AgentManagerClient, repo executorStore, log *logger.Logger, cfg ExecutorConfig) *Executor { return &Executor{ agentManager: agentManager, repo: repo, diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index a108f9e928..089abf2e7c 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -81,7 +81,7 @@ func (m *mockAgentManager) ResolveAgentProfile(ctx context.Context, profileID st }, nil } -// mockRepository implements repository.Repository for testing +// mockRepository implements executorStore for testing type mockRepository struct { sessions map[string]*models.TaskSession taskRepositories map[string]*models.TaskRepository @@ -181,7 +181,7 @@ func (m *mockRepository) GetWorkspace(ctx context.Context, id string) (*models.W return nil, nil } -// Stub implementations for other repository methods (matching repository.Repository interface) +// Stub implementations for additional repository methods // Workspace operations func (m *mockRepository) CreateWorkspace(ctx context.Context, workspace *models.Workspace) error { diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 1bac8d5b66..e3a5cf1994 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -28,7 +28,6 @@ import ( "github.com/kandev/kandev/internal/orchestrator/watcher" "github.com/kandev/kandev/internal/secrets" "github.com/kandev/kandev/internal/task/models" - "github.com/kandev/kandev/internal/task/repository" wfmodels "github.com/kandev/kandev/internal/workflow/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -99,6 +98,52 @@ type WorktreeRecreator interface { RecreateWorktree(ctx context.Context, req WorktreeRecreateRequest) (*WorktreeRecreateResult, error) } +// repoStore is the repository interface accepted by NewService. +// It covers both the orchestrator's own needs (sessionExecutorStore) and +// the executor package's needs (executor.executorStore). +type repoStore interface { + sessionExecutorStore + // Additional methods needed by executor + UpdateTaskState(ctx context.Context, id string, state v1.TaskState) error + GetPrimaryTaskRepository(ctx context.Context, taskID string) (*models.TaskRepository, error) + CreateTaskSession(ctx context.Context, session *models.TaskSession) error + UpdateTaskSession(ctx context.Context, session *models.TaskSession) error + SetSessionPrimary(ctx context.Context, sessionID string) error + ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) + ListActiveTaskSessionsByTaskID(ctx context.Context, taskID string) ([]*models.TaskSession, error) + CreateTaskSessionWorktree(ctx context.Context, sessionWorktree *models.TaskSessionWorktree) error + GetRepository(ctx context.Context, id string) (*models.Repository, error) + GetExecutorProfile(ctx context.Context, id string) (*models.ExecutorProfile, error) + GetWorkspace(ctx context.Context, id string) (*models.Workspace, error) +} + +// sessionExecutorStore is the minimal repository interface needed by the orchestrator service. +type sessionExecutorStore interface { + // Session + GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) + UpdateTaskSession(ctx context.Context, session *models.TaskSession) error + UpdateTaskSessionState(ctx context.Context, id string, state models.TaskSessionState, errorMessage string) error + ClearSessionExecutionID(ctx context.Context, id string) error + UpdateSessionWorkflowStep(ctx context.Context, sessionID string, stepID string) error + UpdateSessionReviewStatus(ctx context.Context, sessionID string, status string) error + // Executor running state + ListExecutorsRunning(ctx context.Context) ([]*models.ExecutorRunning, error) + UpsertExecutorRunning(ctx context.Context, running *models.ExecutorRunning) error + GetExecutorRunningBySessionID(ctx context.Context, sessionID string) (*models.ExecutorRunning, error) + DeleteExecutorRunningBySessionID(ctx context.Context, sessionID string) error + // Executor + GetExecutor(ctx context.Context, id string) (*models.Executor, error) + // Task + GetTask(ctx context.Context, id string) (*models.Task, error) + UpdateTask(ctx context.Context, task *models.Task) error + // Git snapshots and commits + GetLatestGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) + CreateGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error + CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error + GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) + DeleteSessionCommit(ctx context.Context, id string) error +} + // WorktreeRecreateRequest contains parameters for recreating a worktree. type WorktreeRecreateRequest struct { SessionID string @@ -121,7 +166,7 @@ type Service struct { logger *logger.Logger eventBus bus.EventBus taskRepo scheduler.TaskRepository - repo repository.Repository // Full repository for agent sessions + repo sessionExecutorStore agentManager executor.AgentManagerClient // Components @@ -171,7 +216,7 @@ func NewService( eventBus bus.EventBus, agentManager executor.AgentManagerClient, taskRepo scheduler.TaskRepository, - repo repository.Repository, + repo repoStore, shellPrefs executor.ShellPreferenceProvider, secretStore secrets.SecretStore, log *logger.Logger, diff --git a/apps/backend/internal/task/handlers/message_handlers_test.go b/apps/backend/internal/task/handlers/message_handlers_test.go index d42fc04351..4ad03b400d 100644 --- a/apps/backend/internal/task/handlers/message_handlers_test.go +++ b/apps/backend/internal/task/handlers/message_handlers_test.go @@ -50,7 +50,12 @@ func newTestMessageHandlers(t *testing.T, repo *sessionStateSequencer) *MessageH Format: "json", }) require.NoError(t, err) - svc := service.NewService(repo, nil, log, service.RepositoryDiscoveryConfig{}) + svc := service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, Reviews: repo, + }, nil, log, service.RepositoryDiscoveryConfig{}) return NewMessageHandlers(svc, nil, log) } diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 8de7f91f22..6ac1b0b525 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -425,7 +425,12 @@ func newTestService(t *testing.T, scripts map[string][]*models.RepositoryScript) t.Fatalf("failed to create logger: %v", err) } repo := &mockRepository{scriptsByRepo: scripts} - return service.NewService(repo, nil, log, service.RepositoryDiscoveryConfig{}) + return service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, Reviews: repo, + }, nil, log, service.RepositoryDiscoveryConfig{}) } func addExecution(t *testing.T, mgr *lifecycle.Manager, exec *lifecycle.AgentExecution) { @@ -572,7 +577,12 @@ func TestStopProcessRejectsDifferentSession(t *testing.T) { "exec-1": {ID: "exec-1", Type: models.ExecutorTypeLocal}, }, } - svc := service.NewService(repo, nil, log, service.RepositoryDiscoveryConfig{}) + svc := service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, Reviews: repo, + }, nil, log, service.RepositoryDiscoveryConfig{}) lifecycleMgr := newLifecycleManager(t, log) client := newAgentctlClient(t, server.URL, log) exec := (&lifecycle.ExecutorInstance{ @@ -631,7 +641,12 @@ func TestStopProcessAgentctlUnavailable(t *testing.T) { "exec-1": {ID: "exec-1", Type: models.ExecutorTypeLocal}, }, } - svc := service.NewService(repo, nil, log, service.RepositoryDiscoveryConfig{}) + svc := service.NewService(service.Repos{ + Workspaces: repo, Tasks: repo, TaskRepos: repo, + Workflows: repo, Messages: repo, Turns: repo, + Sessions: repo, GitSnapshots: repo, RepoEntities: repo, + Executors: repo, Environments: repo, Reviews: repo, + }, nil, log, service.RepositoryDiscoveryConfig{}) lifecycleMgr := newLifecycleManager(t, log) client := newAgentctlClient(t, server.URL, log) exec := (&lifecycle.ExecutorInstance{ diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index 2be574716d..b540fbd0ef 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -8,16 +8,24 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/task/dto" - "github.com/kandev/kandev/internal/task/repository" + "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/service" ws "github.com/kandev/kandev/pkg/websocket" "go.uber.org/zap" ) +// handlerRepo is the minimal repository interface needed by task handlers. +type handlerRepo interface { + UpsertSessionFileReview(ctx context.Context, review *models.SessionFileReview) error + GetSessionFileReviews(ctx context.Context, sessionID string) ([]*models.SessionFileReview, error) + DeleteSessionFileReviews(ctx context.Context, sessionID string) error + ListTurnsBySession(ctx context.Context, sessionID string) ([]*models.Turn, error) +} + type TaskHandlers struct { service *service.Service orchestrator OrchestratorStarter - repo repository.Repository + repo handlerRepo planService *service.PlanService logger *logger.Logger } @@ -31,7 +39,7 @@ type OrchestratorStarter interface { StartTaskWithSession(ctx context.Context, taskID string, sessionID string, agentProfileID string, executorID string, executorProfileID string, priority int, prompt string, workflowStepID string, planMode bool) (*executor.TaskExecution, error) } -func NewTaskHandlers(svc *service.Service, orchestrator OrchestratorStarter, repo repository.Repository, planService *service.PlanService, log *logger.Logger) *TaskHandlers { +func NewTaskHandlers(svc *service.Service, orchestrator OrchestratorStarter, repo handlerRepo, planService *service.PlanService, log *logger.Logger) *TaskHandlers { return &TaskHandlers{ service: svc, orchestrator: orchestrator, @@ -41,7 +49,7 @@ func NewTaskHandlers(svc *service.Service, orchestrator OrchestratorStarter, rep } } -func RegisterTaskRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, svc *service.Service, orchestrator OrchestratorStarter, repo repository.Repository, planService *service.PlanService, log *logger.Logger) { +func RegisterTaskRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, svc *service.Service, orchestrator OrchestratorStarter, repo handlerRepo, planService *service.PlanService, log *logger.Logger) { handlers := NewTaskHandlers(svc, orchestrator, repo, planService, log) handlers.registerHTTP(router) handlers.registerWS(dispatcher) diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 47a65a8e94..8012ce2d54 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -7,16 +7,18 @@ import ( v1 "github.com/kandev/kandev/pkg/api/v1" ) -// Repository defines the interface for task storage operations -type Repository interface { - // Workspace operations +// WorkspaceRepository handles workspace CRUD. +type WorkspaceRepository interface { CreateWorkspace(ctx context.Context, workspace *models.Workspace) error GetWorkspace(ctx context.Context, id string) (*models.Workspace, error) UpdateWorkspace(ctx context.Context, workspace *models.Workspace) error DeleteWorkspace(ctx context.Context, id string) error ListWorkspaces(ctx context.Context) ([]*models.Workspace, error) +} - // Task operations +// TaskRepository handles task CRUD and workflow placement. +// Note: models.TaskRepository is a struct in internal/task/models; no Go conflict exists. +type TaskRepository interface { CreateTask(ctx context.Context, task *models.Task) error GetTask(ctx context.Context, id string) (*models.Task, error) UpdateTask(ctx context.Context, task *models.Task) error @@ -31,8 +33,11 @@ type Repository interface { CountTasksByWorkflowStep(ctx context.Context, stepID string) (int, error) AddTaskToWorkflow(ctx context.Context, taskID, workflowID, workflowStepID string, position int) error RemoveTaskFromWorkflow(ctx context.Context, taskID, workflowID string) error +} - // TaskRepository operations +// TaskRepoRepository handles the task↔repository junction table (models.TaskRepository rows). +// Named TaskRepoRepository to reduce reader confusion with the TaskRepository sub-interface above. +type TaskRepoRepository interface { CreateTaskRepository(ctx context.Context, taskRepo *models.TaskRepository) error GetTaskRepository(ctx context.Context, id string) (*models.TaskRepository, error) ListTaskRepositories(ctx context.Context, taskID string) ([]*models.TaskRepository, error) @@ -41,15 +46,19 @@ type Repository interface { DeleteTaskRepository(ctx context.Context, id string) error DeleteTaskRepositoriesByTask(ctx context.Context, taskID string) error GetPrimaryTaskRepository(ctx context.Context, taskID string) (*models.TaskRepository, error) +} - // Workflow operations +// WorkflowRepository handles workflow CRUD. +type WorkflowRepository interface { CreateWorkflow(ctx context.Context, workflow *models.Workflow) error GetWorkflow(ctx context.Context, id string) (*models.Workflow, error) UpdateWorkflow(ctx context.Context, workflow *models.Workflow) error DeleteWorkflow(ctx context.Context, id string) error ListWorkflows(ctx context.Context, workspaceID string) ([]*models.Workflow, error) +} - // Message operations +// MessageRepository handles message persistence and lookups. +type MessageRepository interface { CreateMessage(ctx context.Context, message *models.Message) error GetMessage(ctx context.Context, id string) (*models.Message, error) GetMessageByToolCallID(ctx context.Context, sessionID, toolCallID string) (*models.Message, error) @@ -59,8 +68,10 @@ type Repository interface { ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) ListMessagesPaginated(ctx context.Context, sessionID string, opts models.ListMessagesOptions) ([]*models.Message, bool, error) DeleteMessage(ctx context.Context, id string) error +} - // Turn operations +// TurnRepository handles conversation turn persistence. +type TurnRepository interface { CreateTurn(ctx context.Context, turn *models.Turn) error GetTurn(ctx context.Context, id string) (*models.Turn, error) GetActiveTurnBySessionID(ctx context.Context, sessionID string) (*models.Turn, error) @@ -68,8 +79,10 @@ type Repository interface { CompleteTurn(ctx context.Context, id string) error CompleteRunningToolCallsForTurn(ctx context.Context, turnID string) (int64, error) ListTurnsBySession(ctx context.Context, sessionID string) ([]*models.Turn, error) +} - // Task Session operations +// SessionRepository handles task session lifecycle and workflow-session relationships. +type SessionRepository interface { CreateTaskSession(ctx context.Context, session *models.TaskSession) error GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) GetTaskSessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) @@ -85,8 +98,6 @@ type Repository interface { HasActiveTaskSessionsByEnvironment(ctx context.Context, environmentID string) (bool, error) HasActiveTaskSessionsByRepository(ctx context.Context, repositoryID string) (bool, error) DeleteTaskSession(ctx context.Context, id string) error - - // Workflow-related session operations GetPrimarySessionByTaskID(ctx context.Context, taskID string) (*models.TaskSession, error) GetPrimarySessionIDsByTaskIDs(ctx context.Context, taskIDs []string) (map[string]string, error) GetSessionCountsByTaskIDs(ctx context.Context, taskIDs []string) (map[string]int, error) @@ -94,80 +105,85 @@ type Repository interface { SetSessionPrimary(ctx context.Context, sessionID string) error UpdateSessionWorkflowStep(ctx context.Context, sessionID string, stepID string) error UpdateSessionReviewStatus(ctx context.Context, sessionID string, status string) error +} - // Task Session Worktree operations +// SessionWorktreeRepository handles the task session↔worktree association. +type SessionWorktreeRepository interface { CreateTaskSessionWorktree(ctx context.Context, sessionWorktree *models.TaskSessionWorktree) error ListTaskSessionWorktrees(ctx context.Context, sessionID string) ([]*models.TaskSessionWorktree, error) ListWorktreesBySessionIDs(ctx context.Context, sessionIDs []string) (map[string][]*models.TaskSessionWorktree, error) DeleteTaskSessionWorktree(ctx context.Context, id string) error DeleteTaskSessionWorktreesBySession(ctx context.Context, sessionID string) error +} - // Git Snapshot operations +// GitSnapshotRepository handles git snapshots and session commit records. +type GitSnapshotRepository interface { CreateGitSnapshot(ctx context.Context, snapshot *models.GitSnapshot) error GetLatestGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) GetFirstGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) GetGitSnapshotsBySession(ctx context.Context, sessionID string, limit int) ([]*models.GitSnapshot, error) - - // Session Commit operations CreateSessionCommit(ctx context.Context, commit *models.SessionCommit) error GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) GetLatestSessionCommit(ctx context.Context, sessionID string) (*models.SessionCommit, error) DeleteSessionCommit(ctx context.Context, id string) error +} - // Repository operations +// RepositoryEntityRepository handles git repository entity CRUD and repository scripts. +// Named RepositoryEntityRepository to avoid conflation with the Repository interface itself; +// mirrors the sqlite/repository_entity.go implementation file. +type RepositoryEntityRepository interface { CreateRepository(ctx context.Context, repository *models.Repository) error GetRepository(ctx context.Context, id string) (*models.Repository, error) UpdateRepository(ctx context.Context, repository *models.Repository) error DeleteRepository(ctx context.Context, id string) error ListRepositories(ctx context.Context, workspaceID string) ([]*models.Repository, error) - - // Repository script operations CreateRepositoryScript(ctx context.Context, script *models.RepositoryScript) error GetRepositoryScript(ctx context.Context, id string) (*models.RepositoryScript, error) UpdateRepositoryScript(ctx context.Context, script *models.RepositoryScript) error DeleteRepositoryScript(ctx context.Context, id string) error ListRepositoryScripts(ctx context.Context, repositoryID string) ([]*models.RepositoryScript, error) ListScriptsByRepositoryIDs(ctx context.Context, repoIDs []string) (map[string][]*models.RepositoryScript, error) +} - // Executor operations +// ExecutorRepository handles executor CRUD, executor profiles, and running state. +type ExecutorRepository interface { CreateExecutor(ctx context.Context, executor *models.Executor) error GetExecutor(ctx context.Context, id string) (*models.Executor, error) UpdateExecutor(ctx context.Context, executor *models.Executor) error DeleteExecutor(ctx context.Context, id string) error ListExecutors(ctx context.Context) ([]*models.Executor, error) - - // Executor profile operations CreateExecutorProfile(ctx context.Context, profile *models.ExecutorProfile) error GetExecutorProfile(ctx context.Context, id string) (*models.ExecutorProfile, error) UpdateExecutorProfile(ctx context.Context, profile *models.ExecutorProfile) error DeleteExecutorProfile(ctx context.Context, id string) error ListExecutorProfiles(ctx context.Context, executorID string) ([]*models.ExecutorProfile, error) ListAllExecutorProfiles(ctx context.Context) ([]*models.ExecutorProfile, error) - - // Executor running operations ListExecutorsRunning(ctx context.Context) ([]*models.ExecutorRunning, error) UpsertExecutorRunning(ctx context.Context, running *models.ExecutorRunning) error GetExecutorRunningBySessionID(ctx context.Context, sessionID string) (*models.ExecutorRunning, error) DeleteExecutorRunningBySessionID(ctx context.Context, sessionID string) error +} - // Environment operations +// EnvironmentRepository handles environment CRUD. +type EnvironmentRepository interface { CreateEnvironment(ctx context.Context, environment *models.Environment) error GetEnvironment(ctx context.Context, id string) (*models.Environment, error) UpdateEnvironment(ctx context.Context, environment *models.Environment) error DeleteEnvironment(ctx context.Context, id string) error ListEnvironments(ctx context.Context) ([]*models.Environment, error) +} - // Session File Review operations +// ReviewRepository handles session file review records. +type ReviewRepository interface { UpsertSessionFileReview(ctx context.Context, review *models.SessionFileReview) error GetSessionFileReviews(ctx context.Context, sessionID string) ([]*models.SessionFileReview, error) DeleteSessionFileReviews(ctx context.Context, sessionID string) error +} - // Task Plan operations +// PlanRepository handles task plan CRUD. +type PlanRepository interface { CreateTaskPlan(ctx context.Context, plan *models.TaskPlan) error GetTaskPlan(ctx context.Context, taskID string) (*models.TaskPlan, error) UpdateTaskPlan(ctx context.Context, plan *models.TaskPlan) error DeleteTaskPlan(ctx context.Context, taskID string) error - - // Close closes the repository (for database connections) - Close() error } diff --git a/apps/backend/internal/task/repository/provider.go b/apps/backend/internal/task/repository/provider.go index 10d0a85d7f..3e38aebdd4 100644 --- a/apps/backend/internal/task/repository/provider.go +++ b/apps/backend/internal/task/repository/provider.go @@ -7,7 +7,7 @@ import ( ) // Provide creates the SQLite repository using separate writer and reader pools. -func Provide(writer, reader *sqlx.DB) (Repository, func() error, error) { +func Provide(writer, reader *sqlx.DB) (*sqlite.Repository, func() error, error) { repo, err := sqlite.NewWithDB(writer, reader) if err != nil { return nil, nil, err diff --git a/apps/backend/internal/task/service/auto_archive.go b/apps/backend/internal/task/service/auto_archive.go index d675037934..4feac4a892 100644 --- a/apps/backend/internal/task/service/auto_archive.go +++ b/apps/backend/internal/task/service/auto_archive.go @@ -26,7 +26,7 @@ func (s *Service) StartAutoArchiveLoop(ctx context.Context) { } func (s *Service) runAutoArchive(ctx context.Context) { - tasks, err := s.repo.ListTasksForAutoArchive(ctx) + tasks, err := s.tasks.ListTasksForAutoArchive(ctx) if err != nil { s.logger.Error("auto-archive: failed to list candidates", zap.Error(err)) return diff --git a/apps/backend/internal/task/service/plan_service.go b/apps/backend/internal/task/service/plan_service.go index 17cd864bf3..e51ccf91d3 100644 --- a/apps/backend/internal/task/service/plan_service.go +++ b/apps/backend/internal/task/service/plan_service.go @@ -24,13 +24,13 @@ const createdByAgent = "agent" // PlanService provides task plan business logic. type PlanService struct { - repo repository.Repository + repo repository.PlanRepository eventBus bus.EventBus logger *logger.Logger } // NewPlanService creates a new task plan service. -func NewPlanService(repo repository.Repository, eventBus bus.EventBus, log *logger.Logger) *PlanService { +func NewPlanService(repo repository.PlanRepository, eventBus bus.EventBus, log *logger.Logger) *PlanService { return &PlanService{ repo: repo, eventBus: eventBus, diff --git a/apps/backend/internal/task/service/repository_discovery.go b/apps/backend/internal/task/service/repository_discovery.go index 4631e572c0..c7d49e5615 100644 --- a/apps/backend/internal/task/service/repository_discovery.go +++ b/apps/backend/internal/task/service/repository_discovery.go @@ -128,7 +128,7 @@ func (s *Service) ValidateLocalRepositoryPath(ctx context.Context, path string) } func (s *Service) ListRepositoryBranches(ctx context.Context, repoID string) ([]Branch, error) { - repo, err := s.repo.GetRepository(ctx, repoID) + repo, err := s.repoEntities.GetRepository(ctx, repoID) if err != nil { return nil, err } diff --git a/apps/backend/internal/task/service/repository_discovery_test.go b/apps/backend/internal/task/service/repository_discovery_test.go index adcbc4c5c9..add28acbdc 100644 --- a/apps/backend/internal/task/service/repository_discovery_test.go +++ b/apps/backend/internal/task/service/repository_discovery_test.go @@ -295,7 +295,20 @@ func newDiscoveryService(t *testing.T, root string) *Service { }) log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"}) eventBus := bus.NewMemoryEventBus(log) - return NewService(repo, eventBus, log, RepositoryDiscoveryConfig{ + return NewService(Repos{ + Workspaces: repo, + Tasks: repo, + TaskRepos: repo, + Workflows: repo, + Messages: repo, + Turns: repo, + Sessions: repo, + GitSnapshots: repo, + RepoEntities: repo, + Executors: repo, + Environments: repo, + Reviews: repo, + }, eventBus, log, RepositoryDiscoveryConfig{ Roots: []string{root}, MaxDepth: 6, }) diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 9e4c18c345..0abff8cd54 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -83,9 +83,36 @@ func validateExecutorConfig(config map[string]string) error { return nil } +// Repos holds the repository sub-interfaces used by the task service. +type Repos struct { + Workspaces repository.WorkspaceRepository + Tasks repository.TaskRepository + TaskRepos repository.TaskRepoRepository + Workflows repository.WorkflowRepository + Messages repository.MessageRepository + Turns repository.TurnRepository + Sessions repository.SessionRepository + GitSnapshots repository.GitSnapshotRepository + RepoEntities repository.RepositoryEntityRepository + Executors repository.ExecutorRepository + Environments repository.EnvironmentRepository + Reviews repository.ReviewRepository +} + // Service provides task business logic type Service struct { - repo repository.Repository + workspaces repository.WorkspaceRepository + tasks repository.TaskRepository + taskRepos repository.TaskRepoRepository + workflows repository.WorkflowRepository + messages repository.MessageRepository + turns repository.TurnRepository + sessions repository.SessionRepository + gitSnapshots repository.GitSnapshotRepository + repoEntities repository.RepositoryEntityRepository + executors repository.ExecutorRepository + environments repository.EnvironmentRepository + reviews repository.ReviewRepository eventBus bus.EventBus logger *logger.Logger discoveryConfig RepositoryDiscoveryConfig @@ -97,9 +124,20 @@ type Service struct { } // NewService creates a new task service -func NewService(repo repository.Repository, eventBus bus.EventBus, log *logger.Logger, discoveryConfig RepositoryDiscoveryConfig) *Service { +func NewService(repos Repos, eventBus bus.EventBus, log *logger.Logger, discoveryConfig RepositoryDiscoveryConfig) *Service { return &Service{ - repo: repo, + workspaces: repos.Workspaces, + tasks: repos.Tasks, + taskRepos: repos.TaskRepos, + workflows: repos.Workflows, + messages: repos.Messages, + turns: repos.Turns, + sessions: repos.Sessions, + gitSnapshots: repos.GitSnapshots, + repoEntities: repos.RepoEntities, + executors: repos.Executors, + environments: repos.Environments, + reviews: repos.Reviews, eventBus: eventBus, logger: log, discoveryConfig: discoveryConfig, diff --git a/apps/backend/internal/task/service/service_messages.go b/apps/backend/internal/task/service/service_messages.go index bc84a4e5fb..df12ca09be 100644 --- a/apps/backend/internal/task/service/service_messages.go +++ b/apps/backend/internal/task/service/service_messages.go @@ -14,7 +14,7 @@ import ( // CreateMessage creates a new message on an agent session func (s *Service) CreateMessage(ctx context.Context, req *CreateMessageRequest) (*models.Message, error) { - session, err := s.repo.GetTaskSession(ctx, req.TaskSessionID) + session, err := s.sessions.GetTaskSession(ctx, req.TaskSessionID) if err != nil { return nil, err } @@ -62,7 +62,7 @@ func (s *Service) CreateMessage(ctx context.Context, req *CreateMessageRequest) CreatedAt: time.Now().UTC(), } - if err := s.repo.CreateMessage(ctx, message); err != nil { + if err := s.messages.CreateMessage(ctx, message); err != nil { s.logger.Error("failed to create message", zap.Error(err)) return nil, err } @@ -113,7 +113,7 @@ func (s *Service) getSessionWithRetry(ctx context.Context, sessionID, messageID var session *models.TaskSession var err error for attempt := 0; attempt < maxRetries; attempt++ { - session, err = s.repo.GetTaskSession(ctx, sessionID) + session, err = s.sessions.GetTaskSession(ctx, sessionID) if err == nil { return session, nil } @@ -184,7 +184,7 @@ func (s *Service) buildMessage(ctx context.Context, id string, req *CreateMessag func (s *Service) createMessageWithRetry(ctx context.Context, message *models.Message, maxRetries int, retryDelay time.Duration) error { var err error for attempt := 0; attempt < maxRetries; attempt++ { - err = s.repo.CreateMessage(ctx, message) + err = s.messages.CreateMessage(ctx, message) if err == nil { return nil } @@ -209,12 +209,12 @@ func (s *Service) createMessageWithRetry(ctx context.Context, message *models.Me // GetMessage retrieves a message by ID func (s *Service) GetMessage(ctx context.Context, id string) (*models.Message, error) { - return s.repo.GetMessage(ctx, id) + return s.messages.GetMessage(ctx, id) } // ListMessages returns all messages for a session. func (s *Service) ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) { - return s.repo.ListMessages(ctx, sessionID) + return s.messages.ListMessages(ctx, sessionID) } // ListMessagesPaginated returns messages for a session with pagination options. @@ -226,7 +226,7 @@ func (s *Service) ListMessagesPaginated(ctx context.Context, req ListMessagesReq if limit > MaxMessagesPageSize { limit = MaxMessagesPageSize } - return s.repo.ListMessagesPaginated(ctx, req.TaskSessionID, models.ListMessagesOptions{ + return s.messages.ListMessagesPaginated(ctx, req.TaskSessionID, models.ListMessagesOptions{ Limit: limit, Before: req.Before, After: req.After, @@ -236,7 +236,7 @@ func (s *Service) ListMessagesPaginated(ctx context.Context, req ListMessagesReq // DeleteMessage deletes a message func (s *Service) DeleteMessage(ctx context.Context, id string) error { - if err := s.repo.DeleteMessage(ctx, id); err != nil { + if err := s.messages.DeleteMessage(ctx, id); err != nil { s.logger.Error("failed to delete message", zap.String("message_id", id), zap.Error(err)) return err } @@ -247,7 +247,7 @@ func (s *Service) DeleteMessage(ctx context.Context, id string) error { // UpdateMessage updates an existing message and publishes an event. func (s *Service) UpdateMessage(ctx context.Context, message *models.Message) error { - if err := s.repo.UpdateMessage(ctx, message); err != nil { + if err := s.messages.UpdateMessage(ctx, message); err != nil { s.logger.Error("failed to update message", zap.String("message_id", message.ID), zap.Error(err)) @@ -263,7 +263,7 @@ func (s *Service) UpdateMessage(ctx context.Context, message *models.Message) er // AppendMessageContent appends additional content to an existing message. // This is used for streaming agent responses where content arrives incrementally. func (s *Service) AppendMessageContent(ctx context.Context, messageID, additionalContent string) error { - message, err := s.repo.GetMessage(ctx, messageID) + message, err := s.messages.GetMessage(ctx, messageID) if err != nil { s.logger.Warn("message not found for append", zap.String("message_id", messageID), @@ -274,7 +274,7 @@ func (s *Service) AppendMessageContent(ctx context.Context, messageID, additiona // Append the new content message.Content += additionalContent - if err := s.repo.UpdateMessage(ctx, message); err != nil { + if err := s.messages.UpdateMessage(ctx, message); err != nil { s.logger.Error("failed to append message content", zap.String("message_id", messageID), zap.Error(err)) @@ -295,7 +295,7 @@ func (s *Service) AppendMessageContent(ctx context.Context, messageID, additiona // AppendThinkingContent appends additional thinking content to an existing thinking message. // This updates the metadata.thinking field for streaming agent reasoning. func (s *Service) AppendThinkingContent(ctx context.Context, messageID, additionalContent string) error { - message, err := s.repo.GetMessage(ctx, messageID) + message, err := s.messages.GetMessage(ctx, messageID) if err != nil { s.logger.Warn("thinking message not found for append", zap.String("message_id", messageID), @@ -315,7 +315,7 @@ func (s *Service) AppendThinkingContent(ctx context.Context, messageID, addition } message.Metadata["thinking"] = existingThinking + additionalContent - if err := s.repo.UpdateMessage(ctx, message); err != nil { + if err := s.messages.UpdateMessage(ctx, message); err != nil { s.logger.Error("failed to append thinking content", zap.String("message_id", messageID), zap.Error(err)) @@ -366,7 +366,7 @@ func (s *Service) UpdateToolCallMessageWithCreate(ctx context.Context, sessionID s.applyToolCallMessageUpdate(message, status, result, title, normalized) - if err := s.repo.UpdateMessage(ctx, message); err != nil { + if err := s.messages.UpdateMessage(ctx, message); err != nil { s.logger.Error("failed to update tool call message", zap.String("message_id", message.ID), zap.String("tool_call_id", toolCallID), @@ -390,7 +390,7 @@ func (s *Service) getToolCallMessageWithRetry(ctx context.Context, sessionID, to var message *models.Message var err error for attempt := 0; attempt < maxRetries; attempt++ { - message, err = s.repo.GetMessageByToolCallID(ctx, sessionID, toolCallID) + message, err = s.messages.GetMessageByToolCallID(ctx, sessionID, toolCallID) if err == nil { return message, nil } @@ -496,7 +496,7 @@ func (s *Service) UpdatePermissionMessage(ctx context.Context, sessionID, pendin // Retry loop to handle race condition for attempt := 0; attempt < maxRetries; attempt++ { - message, err = s.repo.GetMessageByPendingID(ctx, sessionID, pendingID) + message, err = s.messages.GetMessageByPendingID(ctx, sessionID, pendingID) if err == nil { break } @@ -529,7 +529,7 @@ func (s *Service) UpdatePermissionMessage(ctx context.Context, sessionID, pendin } message.Metadata["status"] = status - if err := s.repo.UpdateMessage(ctx, message); err != nil { + if err := s.messages.UpdateMessage(ctx, message); err != nil { s.logger.Error("failed to update permission message", zap.String("message_id", message.ID), zap.String("pending_id", pendingID), @@ -573,7 +573,7 @@ func (s *Service) UpdateClarificationMessage(ctx context.Context, sessionID, pen // Retry loop to handle race condition for attempt := 0; attempt < maxRetries; attempt++ { - message, err = s.repo.GetMessageByPendingID(ctx, sessionID, pendingID) + message, err = s.messages.GetMessageByPendingID(ctx, sessionID, pendingID) if err == nil { break } @@ -609,7 +609,7 @@ func (s *Service) UpdateClarificationMessage(ctx context.Context, sessionID, pen message.Metadata["response"] = answers } - if err := s.repo.UpdateMessage(ctx, message); err != nil { + if err := s.messages.UpdateMessage(ctx, message); err != nil { s.logger.Error("failed to update clarification message", zap.String("message_id", message.ID), zap.String("pending_id", pendingID), diff --git a/apps/backend/internal/task/service/service_resources.go b/apps/backend/internal/task/service/service_resources.go index ea21efdc5c..426a07ca94 100644 --- a/apps/backend/internal/task/service/service_resources.go +++ b/apps/backend/internal/task/service/service_resources.go @@ -28,7 +28,7 @@ func (s *Service) CreateWorkspace(ctx context.Context, req *CreateWorkspaceReque DefaultAgentProfileID: normalizeOptionalID(req.DefaultAgentProfileID), } - if err := s.repo.CreateWorkspace(ctx, workspace); err != nil { + if err := s.workspaces.CreateWorkspace(ctx, workspace); err != nil { s.logger.Error("failed to create workspace", zap.Error(err)) return nil, err } @@ -40,12 +40,12 @@ func (s *Service) CreateWorkspace(ctx context.Context, req *CreateWorkspaceReque // GetWorkspace retrieves a workspace by ID func (s *Service) GetWorkspace(ctx context.Context, id string) (*models.Workspace, error) { - return s.repo.GetWorkspace(ctx, id) + return s.workspaces.GetWorkspace(ctx, id) } // UpdateWorkspace updates an existing workspace func (s *Service) UpdateWorkspace(ctx context.Context, id string, req *UpdateWorkspaceRequest) (*models.Workspace, error) { - workspace, err := s.repo.GetWorkspace(ctx, id) + workspace, err := s.workspaces.GetWorkspace(ctx, id) if err != nil { return nil, err } @@ -67,7 +67,7 @@ func (s *Service) UpdateWorkspace(ctx context.Context, id string, req *UpdateWor } workspace.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateWorkspace(ctx, workspace); err != nil { + if err := s.workspaces.UpdateWorkspace(ctx, workspace); err != nil { s.logger.Error("failed to update workspace", zap.String("workspace_id", id), zap.Error(err)) return nil, err } @@ -79,11 +79,11 @@ func (s *Service) UpdateWorkspace(ctx context.Context, id string, req *UpdateWor // DeleteWorkspace deletes a workspace func (s *Service) DeleteWorkspace(ctx context.Context, id string) error { - workspace, err := s.repo.GetWorkspace(ctx, id) + workspace, err := s.workspaces.GetWorkspace(ctx, id) if err != nil { return err } - if err := s.repo.DeleteWorkspace(ctx, id); err != nil { + if err := s.workspaces.DeleteWorkspace(ctx, id); err != nil { s.logger.Error("failed to delete workspace", zap.String("workspace_id", id), zap.Error(err)) return err } @@ -105,7 +105,7 @@ func normalizeOptionalID(value *string) *string { // ListWorkspaces returns all workspaces func (s *Service) ListWorkspaces(ctx context.Context) ([]*models.Workspace, error) { - return s.repo.ListWorkspaces(ctx) + return s.workspaces.ListWorkspaces(ctx) } // Workflow operations @@ -120,7 +120,7 @@ func (s *Service) CreateWorkflow(ctx context.Context, req *CreateWorkflowRequest WorkflowTemplateID: req.WorkflowTemplateID, } - if err := s.repo.CreateWorkflow(ctx, workflow); err != nil { + if err := s.workflows.CreateWorkflow(ctx, workflow); err != nil { s.logger.Error("failed to create workflow", zap.Error(err)) return nil, err } @@ -143,12 +143,12 @@ func (s *Service) CreateWorkflow(ctx context.Context, req *CreateWorkflowRequest // GetWorkflow retrieves a workflow by ID func (s *Service) GetWorkflow(ctx context.Context, id string) (*models.Workflow, error) { - return s.repo.GetWorkflow(ctx, id) + return s.workflows.GetWorkflow(ctx, id) } // UpdateWorkflow updates an existing workflow func (s *Service) UpdateWorkflow(ctx context.Context, id string, req *UpdateWorkflowRequest) (*models.Workflow, error) { - workflow, err := s.repo.GetWorkflow(ctx, id) + workflow, err := s.workflows.GetWorkflow(ctx, id) if err != nil { return nil, err } @@ -161,7 +161,7 @@ func (s *Service) UpdateWorkflow(ctx context.Context, id string, req *UpdateWork } workflow.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateWorkflow(ctx, workflow); err != nil { + if err := s.workflows.UpdateWorkflow(ctx, workflow); err != nil { s.logger.Error("failed to update workflow", zap.String("workflow_id", id), zap.Error(err)) return nil, err } @@ -173,11 +173,11 @@ func (s *Service) UpdateWorkflow(ctx context.Context, id string, req *UpdateWork // DeleteWorkflow deletes a workflow func (s *Service) DeleteWorkflow(ctx context.Context, id string) error { - workflow, err := s.repo.GetWorkflow(ctx, id) + workflow, err := s.workflows.GetWorkflow(ctx, id) if err != nil { return err } - if err := s.repo.DeleteWorkflow(ctx, id); err != nil { + if err := s.workflows.DeleteWorkflow(ctx, id); err != nil { s.logger.Error("failed to delete workflow", zap.String("workflow_id", id), zap.Error(err)) return err } @@ -189,7 +189,7 @@ func (s *Service) DeleteWorkflow(ctx context.Context, id string) error { // ListWorkflows returns all workflows for a workspace (or all if empty) func (s *Service) ListWorkflows(ctx context.Context, workspaceID string) ([]*models.Workflow, error) { - return s.repo.ListWorkflows(ctx, workspaceID) + return s.workflows.ListWorkflows(ctx, workspaceID) } // Repository operations @@ -228,7 +228,7 @@ func (s *Service) CreateRepository(ctx context.Context, req *CreateRepositoryReq DevScript: req.DevScript, } - if err := s.repo.CreateRepository(ctx, repository); err != nil { + if err := s.repoEntities.CreateRepository(ctx, repository); err != nil { s.logger.Error("failed to create repository", zap.Error(err)) return nil, err } @@ -239,11 +239,11 @@ func (s *Service) CreateRepository(ctx context.Context, req *CreateRepositoryReq } func (s *Service) GetRepository(ctx context.Context, id string) (*models.Repository, error) { - return s.repo.GetRepository(ctx, id) + return s.repoEntities.GetRepository(ctx, id) } func (s *Service) UpdateRepository(ctx context.Context, id string, req *UpdateRepositoryRequest) (*models.Repository, error) { - repository, err := s.repo.GetRepository(ctx, id) + repository, err := s.repoEntities.GetRepository(ctx, id) if err != nil { return nil, err } @@ -252,7 +252,7 @@ func (s *Service) UpdateRepository(ctx context.Context, id string, req *UpdateRe } repository.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateRepository(ctx, repository); err != nil { + if err := s.repoEntities.UpdateRepository(ctx, repository); err != nil { s.logger.Error("failed to update repository", zap.String("repository_id", id), zap.Error(err)) return nil, err } @@ -312,11 +312,11 @@ func applyRepositoryUpdates(repository *models.Repository, req *UpdateRepository } func (s *Service) DeleteRepository(ctx context.Context, id string) error { - repository, err := s.repo.GetRepository(ctx, id) + repository, err := s.repoEntities.GetRepository(ctx, id) if err != nil { return err } - active, err := s.repo.HasActiveTaskSessionsByRepository(ctx, id) + active, err := s.sessions.HasActiveTaskSessionsByRepository(ctx, id) if err != nil { s.logger.Error("failed to check active agent sessions for repository", zap.String("repository_id", id), zap.Error(err)) return err @@ -324,7 +324,7 @@ func (s *Service) DeleteRepository(ctx context.Context, id string) error { if active { return ErrActiveTaskSessions } - if err := s.repo.DeleteRepository(ctx, id); err != nil { + if err := s.repoEntities.DeleteRepository(ctx, id); err != nil { s.logger.Error("failed to delete repository", zap.String("repository_id", id), zap.Error(err)) return err } @@ -334,7 +334,7 @@ func (s *Service) DeleteRepository(ctx context.Context, id string) error { } func (s *Service) ListRepositories(ctx context.Context, workspaceID string) ([]*models.Repository, error) { - return s.repo.ListRepositories(ctx, workspaceID) + return s.repoEntities.ListRepositories(ctx, workspaceID) } // Repository script operations @@ -347,7 +347,7 @@ func (s *Service) CreateRepositoryScript(ctx context.Context, req *CreateReposit Command: req.Command, Position: req.Position, } - if err := s.repo.CreateRepositoryScript(ctx, script); err != nil { + if err := s.repoEntities.CreateRepositoryScript(ctx, script); err != nil { s.logger.Error("failed to create repository script", zap.Error(err)) return nil, err } @@ -357,11 +357,11 @@ func (s *Service) CreateRepositoryScript(ctx context.Context, req *CreateReposit } func (s *Service) GetRepositoryScript(ctx context.Context, id string) (*models.RepositoryScript, error) { - return s.repo.GetRepositoryScript(ctx, id) + return s.repoEntities.GetRepositoryScript(ctx, id) } func (s *Service) UpdateRepositoryScript(ctx context.Context, id string, req *UpdateRepositoryScriptRequest) (*models.RepositoryScript, error) { - script, err := s.repo.GetRepositoryScript(ctx, id) + script, err := s.repoEntities.GetRepositoryScript(ctx, id) if err != nil { return nil, err } @@ -376,7 +376,7 @@ func (s *Service) UpdateRepositoryScript(ctx context.Context, id string, req *Up } script.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateRepositoryScript(ctx, script); err != nil { + if err := s.repoEntities.UpdateRepositoryScript(ctx, script); err != nil { s.logger.Error("failed to update repository script", zap.String("script_id", id), zap.Error(err)) return nil, err } @@ -386,11 +386,11 @@ func (s *Service) UpdateRepositoryScript(ctx context.Context, id string, req *Up } func (s *Service) DeleteRepositoryScript(ctx context.Context, id string) error { - script, err := s.repo.GetRepositoryScript(ctx, id) + script, err := s.repoEntities.GetRepositoryScript(ctx, id) if err != nil { return err } - if err := s.repo.DeleteRepositoryScript(ctx, id); err != nil { + if err := s.repoEntities.DeleteRepositoryScript(ctx, id); err != nil { s.logger.Error("failed to delete repository script", zap.String("script_id", id), zap.Error(err)) return err } @@ -400,12 +400,12 @@ func (s *Service) DeleteRepositoryScript(ctx context.Context, id string) error { } func (s *Service) ListRepositoryScripts(ctx context.Context, repositoryID string) ([]*models.RepositoryScript, error) { - return s.repo.ListRepositoryScripts(ctx, repositoryID) + return s.repoEntities.ListRepositoryScripts(ctx, repositoryID) } // ListScriptsByRepositoryIDs returns scripts for multiple repositories in a single query. func (s *Service) ListScriptsByRepositoryIDs(ctx context.Context, repoIDs []string) (map[string][]*models.RepositoryScript, error) { - return s.repo.ListScriptsByRepositoryIDs(ctx, repoIDs) + return s.repoEntities.ListScriptsByRepositoryIDs(ctx, repoIDs) } // Executor operations @@ -424,7 +424,7 @@ func (s *Service) CreateExecutor(ctx context.Context, req *CreateExecutorRequest Config: req.Config, } - if err := s.repo.CreateExecutor(ctx, executor); err != nil { + if err := s.executors.CreateExecutor(ctx, executor); err != nil { return nil, err } s.publishExecutorEvent(ctx, events.ExecutorCreated, executor) @@ -432,11 +432,11 @@ func (s *Service) CreateExecutor(ctx context.Context, req *CreateExecutorRequest } func (s *Service) GetExecutor(ctx context.Context, id string) (*models.Executor, error) { - return s.repo.GetExecutor(ctx, id) + return s.executors.GetExecutor(ctx, id) } func (s *Service) UpdateExecutor(ctx context.Context, id string, req *UpdateExecutorRequest) (*models.Executor, error) { - executor, err := s.repo.GetExecutor(ctx, id) + executor, err := s.executors.GetExecutor(ctx, id) if err != nil { return nil, err } @@ -445,7 +445,7 @@ func (s *Service) UpdateExecutor(ctx context.Context, id string, req *UpdateExec } applyExecutorUpdates(executor, req) executor.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateExecutor(ctx, executor); err != nil { + if err := s.executors.UpdateExecutor(ctx, executor); err != nil { return nil, err } s.publishExecutorEvent(ctx, events.ExecutorUpdated, executor) @@ -497,14 +497,14 @@ func applyExecutorUpdates(executor *models.Executor, req *UpdateExecutorRequest) } func (s *Service) DeleteExecutor(ctx context.Context, id string) error { - executor, err := s.repo.GetExecutor(ctx, id) + executor, err := s.executors.GetExecutor(ctx, id) if err != nil { return err } if executor.IsSystem { return fmt.Errorf("system executors cannot be deleted") } - active, err := s.repo.HasActiveTaskSessionsByExecutor(ctx, id) + active, err := s.sessions.HasActiveTaskSessionsByExecutor(ctx, id) if err != nil { s.logger.Error("failed to check active agent sessions for executor", zap.String("executor_id", id), zap.Error(err)) return err @@ -512,7 +512,7 @@ func (s *Service) DeleteExecutor(ctx context.Context, id string) error { if active { return ErrActiveTaskSessions } - if err := s.repo.DeleteExecutor(ctx, id); err != nil { + if err := s.executors.DeleteExecutor(ctx, id); err != nil { return err } s.publishExecutorEvent(ctx, events.ExecutorDeleted, executor) @@ -520,7 +520,7 @@ func (s *Service) DeleteExecutor(ctx context.Context, id string) error { } func (s *Service) ListExecutors(ctx context.Context) ([]*models.Executor, error) { - return s.repo.ListExecutors(ctx) + return s.executors.ListExecutors(ctx) } // Executor Profile operations @@ -533,7 +533,7 @@ func (s *Service) CreateExecutorProfile(ctx context.Context, req *CreateExecutor return nil, fmt.Errorf("executor_id is required") } // Verify executor exists - if _, err := s.repo.GetExecutor(ctx, req.ExecutorID); err != nil { + if _, err := s.executors.GetExecutor(ctx, req.ExecutorID); err != nil { return nil, fmt.Errorf("executor not found: %w", err) } profile := &models.ExecutorProfile{ @@ -545,7 +545,7 @@ func (s *Service) CreateExecutorProfile(ctx context.Context, req *CreateExecutor CleanupScript: req.CleanupScript, EnvVars: req.EnvVars, } - if err := s.repo.CreateExecutorProfile(ctx, profile); err != nil { + if err := s.executors.CreateExecutorProfile(ctx, profile); err != nil { return nil, err } s.publishExecutorProfileEvent(ctx, events.ExecutorProfileCreated, profile) @@ -553,11 +553,11 @@ func (s *Service) CreateExecutorProfile(ctx context.Context, req *CreateExecutor } func (s *Service) GetExecutorProfile(ctx context.Context, id string) (*models.ExecutorProfile, error) { - return s.repo.GetExecutorProfile(ctx, id) + return s.executors.GetExecutorProfile(ctx, id) } func (s *Service) UpdateExecutorProfile(ctx context.Context, id string, req *UpdateExecutorProfileRequest) (*models.ExecutorProfile, error) { - profile, err := s.repo.GetExecutorProfile(ctx, id) + profile, err := s.executors.GetExecutorProfile(ctx, id) if err != nil { return nil, err } @@ -579,7 +579,7 @@ func (s *Service) UpdateExecutorProfile(ctx context.Context, id string, req *Upd if req.EnvVars != nil { profile.EnvVars = req.EnvVars } - if err := s.repo.UpdateExecutorProfile(ctx, profile); err != nil { + if err := s.executors.UpdateExecutorProfile(ctx, profile); err != nil { return nil, err } s.publishExecutorProfileEvent(ctx, events.ExecutorProfileUpdated, profile) @@ -587,11 +587,11 @@ func (s *Service) UpdateExecutorProfile(ctx context.Context, id string, req *Upd } func (s *Service) DeleteExecutorProfile(ctx context.Context, id string) error { - profile, err := s.repo.GetExecutorProfile(ctx, id) + profile, err := s.executors.GetExecutorProfile(ctx, id) if err != nil { return err } - if err := s.repo.DeleteExecutorProfile(ctx, id); err != nil { + if err := s.executors.DeleteExecutorProfile(ctx, id); err != nil { return err } s.publishExecutorProfileEvent(ctx, events.ExecutorProfileDeleted, profile) @@ -599,11 +599,11 @@ func (s *Service) DeleteExecutorProfile(ctx context.Context, id string) error { } func (s *Service) ListExecutorProfiles(ctx context.Context, executorID string) ([]*models.ExecutorProfile, error) { - return s.repo.ListExecutorProfiles(ctx, executorID) + return s.executors.ListExecutorProfiles(ctx, executorID) } func (s *Service) ListAllExecutorProfiles(ctx context.Context) ([]*models.ExecutorProfile, error) { - return s.repo.ListAllExecutorProfiles(ctx) + return s.executors.ListAllExecutorProfiles(ctx) } // Environment operations @@ -619,7 +619,7 @@ func (s *Service) CreateEnvironment(ctx context.Context, req *CreateEnvironmentR Dockerfile: req.Dockerfile, BuildConfig: req.BuildConfig, } - if err := s.repo.CreateEnvironment(ctx, environment); err != nil { + if err := s.environments.CreateEnvironment(ctx, environment); err != nil { return nil, err } s.publishEnvironmentEvent(ctx, events.EnvironmentCreated, environment) @@ -627,11 +627,11 @@ func (s *Service) CreateEnvironment(ctx context.Context, req *CreateEnvironmentR } func (s *Service) GetEnvironment(ctx context.Context, id string) (*models.Environment, error) { - return s.repo.GetEnvironment(ctx, id) + return s.environments.GetEnvironment(ctx, id) } func (s *Service) UpdateEnvironment(ctx context.Context, id string, req *UpdateEnvironmentRequest) (*models.Environment, error) { - environment, err := s.repo.GetEnvironment(ctx, id) + environment, err := s.environments.GetEnvironment(ctx, id) if err != nil { return nil, err } @@ -659,7 +659,7 @@ func (s *Service) UpdateEnvironment(ctx context.Context, id string, req *UpdateE environment.BuildConfig = req.BuildConfig } environment.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateEnvironment(ctx, environment); err != nil { + if err := s.environments.UpdateEnvironment(ctx, environment); err != nil { return nil, err } s.publishEnvironmentEvent(ctx, events.EnvironmentUpdated, environment) @@ -667,14 +667,14 @@ func (s *Service) UpdateEnvironment(ctx context.Context, id string, req *UpdateE } func (s *Service) DeleteEnvironment(ctx context.Context, id string) error { - environment, err := s.repo.GetEnvironment(ctx, id) + environment, err := s.environments.GetEnvironment(ctx, id) if err != nil { return err } if environment.IsSystem { return fmt.Errorf("system environments cannot be deleted") } - active, err := s.repo.HasActiveTaskSessionsByEnvironment(ctx, id) + active, err := s.sessions.HasActiveTaskSessionsByEnvironment(ctx, id) if err != nil { s.logger.Error("failed to check active agent sessions for environment", zap.String("environment_id", id), zap.Error(err)) return err @@ -682,7 +682,7 @@ func (s *Service) DeleteEnvironment(ctx context.Context, id string) error { if active { return ErrActiveTaskSessions } - if err := s.repo.DeleteEnvironment(ctx, id); err != nil { + if err := s.environments.DeleteEnvironment(ctx, id); err != nil { return err } s.publishEnvironmentEvent(ctx, events.EnvironmentDeleted, environment) @@ -690,5 +690,5 @@ func (s *Service) DeleteEnvironment(ctx context.Context, id string) error { } func (s *Service) ListEnvironments(ctx context.Context) ([]*models.Environment, error) { - return s.repo.ListEnvironments(ctx) + return s.environments.ListEnvironments(ctx) } diff --git a/apps/backend/internal/task/service/service_sessions.go b/apps/backend/internal/task/service/service_sessions.go index 0af025bb48..bc9e987600 100644 --- a/apps/backend/internal/task/service/service_sessions.go +++ b/apps/backend/internal/task/service/service_sessions.go @@ -10,39 +10,39 @@ import ( // ListTaskSessions returns all sessions for a task. func (s *Service) ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) { - return s.repo.ListTaskSessions(ctx, taskID) + return s.sessions.ListTaskSessions(ctx, taskID) } // GetTaskSession returns a single session by ID. func (s *Service) GetTaskSession(ctx context.Context, sessionID string) (*models.TaskSession, error) { - return s.repo.GetTaskSession(ctx, sessionID) + return s.sessions.GetTaskSession(ctx, sessionID) } // GetPrimarySession returns the primary session for a task. func (s *Service) GetPrimarySession(ctx context.Context, taskID string) (*models.TaskSession, error) { - return s.repo.GetPrimarySessionByTaskID(ctx, taskID) + return s.sessions.GetPrimarySessionByTaskID(ctx, taskID) } // GetPrimarySessionIDsForTasks returns a map of task ID to primary session ID for the given task IDs. // Tasks without a primary session are not included in the result. func (s *Service) GetPrimarySessionIDsForTasks(ctx context.Context, taskIDs []string) (map[string]string, error) { - return s.repo.GetPrimarySessionIDsByTaskIDs(ctx, taskIDs) + return s.sessions.GetPrimarySessionIDsByTaskIDs(ctx, taskIDs) } // GetSessionCountsForTasks returns a map of task ID to session count for the given task IDs. func (s *Service) GetSessionCountsForTasks(ctx context.Context, taskIDs []string) (map[string]int, error) { - return s.repo.GetSessionCountsByTaskIDs(ctx, taskIDs) + return s.sessions.GetSessionCountsByTaskIDs(ctx, taskIDs) } // GetPrimarySessionInfoForTasks returns a map of task ID to primary session info for the given task IDs. func (s *Service) GetPrimarySessionInfoForTasks(ctx context.Context, taskIDs []string) (map[string]*models.TaskSession, error) { - return s.repo.GetPrimarySessionInfoByTaskIDs(ctx, taskIDs) + return s.sessions.GetPrimarySessionInfoByTaskIDs(ctx, taskIDs) } // SetPrimarySession sets a session as the primary session for its task. // This will unset any existing primary session for the same task. func (s *Service) SetPrimarySession(ctx context.Context, sessionID string) error { - if err := s.repo.SetSessionPrimary(ctx, sessionID); err != nil { + if err := s.sessions.SetSessionPrimary(ctx, sessionID); err != nil { s.logger.Error("failed to set primary session", zap.String("session_id", sessionID), zap.Error(err)) @@ -53,7 +53,7 @@ func (s *Service) SetPrimarySession(ctx context.Context, sessionID string) error // MoveSessionToStep moves a session to a different workflow step. func (s *Service) MoveSessionToStep(ctx context.Context, sessionID string, stepID string) error { - if err := s.repo.UpdateSessionWorkflowStep(ctx, sessionID, stepID); err != nil { + if err := s.sessions.UpdateSessionWorkflowStep(ctx, sessionID, stepID); err != nil { s.logger.Error("failed to move session to step", zap.String("session_id", sessionID), zap.String("step_id", stepID), @@ -65,7 +65,7 @@ func (s *Service) MoveSessionToStep(ctx context.Context, sessionID string, stepI // UpdateSessionReviewStatus updates the review status of a session. func (s *Service) UpdateSessionReviewStatus(ctx context.Context, sessionID string, status string) error { - if err := s.repo.UpdateSessionReviewStatus(ctx, sessionID, status); err != nil { + if err := s.sessions.UpdateSessionReviewStatus(ctx, sessionID, status); err != nil { s.logger.Error("failed to update session review status", zap.String("session_id", sessionID), zap.String("status", status), diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index 19425bf933..e6e0ca1c9e 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -50,7 +50,7 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (*mode Metadata: req.Metadata, } - if err := s.repo.CreateTask(ctx, task); err != nil { + if err := s.tasks.CreateTask(ctx, task); err != nil { s.logger.Error("failed to create task", zap.Error(err)) return nil, err } @@ -60,7 +60,7 @@ func (s *Service) CreateTask(ctx context.Context, req *CreateTaskRequest) (*mode } // Load repositories into task for response - repos, err := s.repo.ListTaskRepositories(ctx, task.ID) + repos, err := s.taskRepos.ListTaskRepositories(ctx, task.ID) if err != nil { s.logger.Error("failed to list task repositories", zap.Error(err)) } else { @@ -78,7 +78,7 @@ func (s *Service) createTaskRepositories(ctx context.Context, taskID, workspaceI var repoByPath map[string]*models.Repository for _, repoInput := range repositories { if repoInput.RepositoryID == "" && repoInput.LocalPath != "" { - repos, err := s.repo.ListRepositories(ctx, workspaceID) + repos, err := s.repoEntities.ListRepositories(ctx, workspaceID) if err != nil { s.logger.Error("failed to list repositories", zap.Error(err)) return err @@ -109,7 +109,7 @@ func (s *Service) createTaskRepositories(ctx context.Context, taskID, workspaceI Position: i, Metadata: make(map[string]interface{}), } - if err := s.repo.CreateTaskRepository(ctx, taskRepo); err != nil { + if err := s.taskRepos.CreateTaskRepository(ctx, taskRepo); err != nil { s.logger.Error("failed to create task repository", zap.Error(err)) return err } @@ -159,7 +159,7 @@ func (s *Service) resolveRepoInput(ctx context.Context, workspaceID string, repo // replaceTaskRepositories deletes all existing task-repository associations and recreates them. func (s *Service) replaceTaskRepositories(ctx context.Context, taskID, workspaceID string, repositories []TaskRepositoryInput) error { - if err := s.repo.DeleteTaskRepositoriesByTask(ctx, taskID); err != nil { + if err := s.taskRepos.DeleteTaskRepositoriesByTask(ctx, taskID); err != nil { s.logger.Error("failed to delete task repositories", zap.Error(err)) return err } @@ -168,13 +168,13 @@ func (s *Service) replaceTaskRepositories(ctx context.Context, taskID, workspace // GetTask retrieves a task by ID and populates repositories func (s *Service) GetTask(ctx context.Context, id string) (*models.Task, error) { - task, err := s.repo.GetTask(ctx, id) + task, err := s.tasks.GetTask(ctx, id) if err != nil { return nil, err } // Load task repositories - repos, err := s.repo.ListTaskRepositories(ctx, id) + repos, err := s.taskRepos.ListTaskRepositories(ctx, id) if err != nil { s.logger.Error("failed to list task repositories", zap.Error(err)) } else { @@ -186,7 +186,7 @@ func (s *Service) GetTask(ctx context.Context, id string) (*models.Task, error) // UpdateTask updates an existing task and publishes a task.updated event func (s *Service) UpdateTask(ctx context.Context, id string, req *UpdateTaskRequest) (*models.Task, error) { - task, err := s.repo.GetTask(ctx, id) + task, err := s.tasks.GetTask(ctx, id) if err != nil { return nil, err } @@ -216,7 +216,7 @@ func (s *Service) UpdateTask(ctx context.Context, id string, req *UpdateTaskRequ } task.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateTask(ctx, task); err != nil { + if err := s.tasks.UpdateTask(ctx, task); err != nil { s.logger.Error("failed to update task", zap.String("task_id", id), zap.Error(err)) return nil, err } @@ -229,7 +229,7 @@ func (s *Service) UpdateTask(ctx context.Context, id string, req *UpdateTaskRequ } // Load repositories into task for response - repos, err := s.repo.ListTaskRepositories(ctx, task.ID) + repos, err := s.taskRepos.ListTaskRepositories(ctx, task.ID) if err != nil { s.logger.Error("failed to list task repositories", zap.Error(err)) } else { @@ -252,7 +252,7 @@ func (s *Service) ArchiveTask(ctx context.Context, id string) error { start := time.Now() // 1. Get task and verify it exists - task, err := s.repo.GetTask(ctx, id) + task, err := s.tasks.GetTask(ctx, id) if err != nil { return err } @@ -264,7 +264,7 @@ func (s *Service) ArchiveTask(ctx context.Context, id string) error { // 2. Gather data needed for cleanup BEFORE archive var activeSessionIDs []string if s.executionStopper != nil { - activeSessions, err := s.repo.ListActiveTaskSessionsByTaskID(ctx, id) + activeSessions, err := s.sessions.ListActiveTaskSessionsByTaskID(ctx, id) if err != nil { s.logger.Warn("failed to list active sessions for archive", zap.String("task_id", id), @@ -275,7 +275,7 @@ func (s *Service) ArchiveTask(ctx context.Context, id string) error { } } - sessions, err := s.repo.ListTaskSessions(ctx, id) + sessions, err := s.sessions.ListTaskSessions(ctx, id) if err != nil { s.logger.Warn("failed to list task sessions for archive", zap.String("task_id", id), @@ -295,12 +295,12 @@ func (s *Service) ArchiveTask(ctx context.Context, id string) error { } // 3. Set archived_at in DB - if err := s.repo.ArchiveTask(ctx, id); err != nil { + if err := s.tasks.ArchiveTask(ctx, id); err != nil { return err } // 4. Re-read task for updated archived_at field - task, err = s.repo.GetTask(ctx, id) + task, err = s.tasks.GetTask(ctx, id) if err != nil { return err } @@ -327,13 +327,13 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { start := time.Now() // 1. Get task (sync, fast) - task, err := s.repo.GetTask(ctx, id) + task, err := s.tasks.GetTask(ctx, id) if err != nil { return err } // 2. Gather data needed for cleanup BEFORE delete (sync, fast) - sessions, err := s.repo.ListTaskSessions(ctx, id) + sessions, err := s.sessions.ListTaskSessions(ctx, id) if err != nil { s.logger.Warn("failed to list task sessions for delete", zap.String("task_id", id), @@ -346,7 +346,7 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { // Must query before delete since DB records will be gone var activeSessionIDs []string if s.executionStopper != nil { - activeSessions, err := s.repo.ListActiveTaskSessionsByTaskID(ctx, id) + activeSessions, err := s.sessions.ListActiveTaskSessionsByTaskID(ctx, id) if err != nil { s.logger.Warn("failed to list active sessions for delete", zap.String("task_id", id), @@ -358,7 +358,7 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { } // 4. Delete from DB (sync, fast) - if err := s.repo.DeleteTask(ctx, id); err != nil { + if err := s.tasks.DeleteTask(ctx, id); err != nil { s.logger.Error("failed to delete task", zap.String("task_id", id), zap.Error(err)) return err } @@ -473,7 +473,7 @@ func (s *Service) performTaskCleanup( if session == nil || session.ID == "" { continue } - if err := s.repo.DeleteExecutorRunningBySessionID(ctx, session.ID); err != nil { + if err := s.executors.DeleteExecutorRunningBySessionID(ctx, session.ID); err != nil { s.logger.Debug("failed to delete executor runtime for session", zap.String("task_id", taskID), zap.String("session_id", session.ID), @@ -487,7 +487,7 @@ func (s *Service) performTaskCleanup( // ListTasks returns all tasks for a workflow func (s *Service) ListTasks(ctx context.Context, workflowID string) ([]*models.Task, error) { - tasks, err := s.repo.ListTasks(ctx, workflowID) + tasks, err := s.tasks.ListTasks(ctx, workflowID) if err != nil { return nil, err } @@ -502,7 +502,7 @@ func (s *Service) ListTasks(ctx context.Context, workflowID string) ([]*models.T // ListTasksByWorkspace returns paginated tasks for a workspace with task repositories loaded. // If query is non-empty, filters by task title, description, repository name, or repository path. func (s *Service) ListTasksByWorkspace(ctx context.Context, workspaceID string, query string, page, pageSize int, includeArchived bool) ([]*models.Task, int, error) { - tasks, total, err := s.repo.ListTasksByWorkspace(ctx, workspaceID, query, page, pageSize, includeArchived) + tasks, total, err := s.tasks.ListTasksByWorkspace(ctx, workspaceID, query, page, pageSize, includeArchived) if err != nil { return nil, 0, err } @@ -523,7 +523,7 @@ func (s *Service) loadTaskRepositoriesBatch(ctx context.Context, tasks []*models for i, t := range tasks { taskIDs[i] = t.ID } - repoMap, err := s.repo.ListTaskRepositoriesByTaskIDs(ctx, taskIDs) + repoMap, err := s.taskRepos.ListTaskRepositoriesByTaskIDs(ctx, taskIDs) if err != nil { return err } diff --git a/apps/backend/internal/task/service/service_test.go b/apps/backend/internal/task/service/service_test.go index 2c84a514ab..b23fcd3b73 100644 --- a/apps/backend/internal/task/service/service_test.go +++ b/apps/backend/internal/task/service/service_test.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" "github.com/kandev/kandev/internal/worktree" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -71,7 +72,7 @@ func (m *MockEventBus) ClearEvents() { m.publishedEvents = make([]*bus.Event, 0) } -func createTestService(t *testing.T) (*Service, *MockEventBus, repository.Repository) { +func createTestService(t *testing.T) (*Service, *MockEventBus, *sqliterepo.Repository) { t.Helper() tmpDir := t.TempDir() dbConn, err := db.OpenSQLite(filepath.Join(tmpDir, "test.db")) @@ -79,11 +80,10 @@ func createTestService(t *testing.T) (*Service, *MockEventBus, repository.Reposi t.Fatalf("failed to open test database: %v", err) } sqlxDB := sqlx.NewDb(dbConn, "sqlite3") - repoImpl, cleanup, err := repository.Provide(sqlxDB, sqlxDB) + repo, cleanup, err := repository.Provide(sqlxDB, sqlxDB) if err != nil { t.Fatalf("failed to create test repository: %v", err) } - repo := repoImpl if _, err := worktree.NewSQLiteStore(sqlxDB, sqlxDB); err != nil { t.Fatalf("failed to init worktree store: %v", err) } @@ -97,7 +97,20 @@ func createTestService(t *testing.T) (*Service, *MockEventBus, repository.Reposi }) eventBus := NewMockEventBus() log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"}) - svc := NewService(repo, eventBus, log, RepositoryDiscoveryConfig{}) + svc := NewService(Repos{ + Workspaces: repo, + Tasks: repo, + TaskRepos: repo, + Workflows: repo, + Messages: repo, + Turns: repo, + Sessions: repo, + GitSnapshots: repo, + RepoEntities: repo, + Executors: repo, + Environments: repo, + Reviews: repo, + }, eventBus, log, RepositoryDiscoveryConfig{}) return svc, eventBus, repo } @@ -462,7 +475,7 @@ func TestService_ListWorkflows(t *testing.T) { // Message tests -func setupTestTask(t *testing.T, repo repository.Repository) { +func setupTestTask(t *testing.T, repo *sqliterepo.Repository) { t.Helper() ctx := context.Background() _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) @@ -470,7 +483,7 @@ func setupTestTask(t *testing.T, repo repository.Repository) { _ = repo.CreateTask(ctx, &models.Task{ID: "task-123", WorkspaceID: "ws-1", WorkflowID: "wf-123", WorkflowStepID: "step-123", Title: "Test Task"}) } -func setupTestSession(t *testing.T, repo repository.Repository) string { +func setupTestSession(t *testing.T, repo *sqliterepo.Repository) string { t.Helper() ctx := context.Background() session := &models.TaskSession{ @@ -483,7 +496,7 @@ func setupTestSession(t *testing.T, repo repository.Repository) string { return session.ID } -func setupTestTurn(t *testing.T, repo repository.Repository, sessionID, taskID, turnID string) string { +func setupTestTurn(t *testing.T, repo *sqliterepo.Repository, sessionID, taskID, turnID string) string { t.Helper() ctx := context.Background() now := time.Now() diff --git a/apps/backend/internal/task/service/service_turns.go b/apps/backend/internal/task/service/service_turns.go index 1cdcd2cabb..a46ab41c46 100644 --- a/apps/backend/internal/task/service/service_turns.go +++ b/apps/backend/internal/task/service/service_turns.go @@ -21,7 +21,7 @@ import ( // StartTurn creates a new turn for a session and publishes the turn.started event. // Returns the created turn. func (s *Service) StartTurn(ctx context.Context, sessionID string) (*models.Turn, error) { - session, err := s.repo.GetTaskSession(ctx, sessionID) + session, err := s.sessions.GetTaskSession(ctx, sessionID) if err != nil { return nil, fmt.Errorf("failed to get session: %w", err) } @@ -35,7 +35,7 @@ func (s *Service) StartTurn(ctx context.Context, sessionID string) (*models.Turn UpdatedAt: time.Now().UTC(), } - if err := s.repo.CreateTurn(ctx, turn); err != nil { + if err := s.turns.CreateTurn(ctx, turn); err != nil { s.logger.Error("failed to create turn", zap.Error(err)) return nil, err } @@ -56,13 +56,13 @@ func (s *Service) CompleteTurn(ctx context.Context, turnID string) error { return nil // No active turn to complete } - if err := s.repo.CompleteTurn(ctx, turnID); err != nil { + if err := s.turns.CompleteTurn(ctx, turnID); err != nil { s.logger.Error("failed to complete turn", zap.String("turn_id", turnID), zap.Error(err)) return err } // Safety net: mark any tool calls still in "running" state as "complete" - if affected, err := s.repo.CompleteRunningToolCallsForTurn(ctx, turnID); err != nil { + if affected, err := s.turns.CompleteRunningToolCallsForTurn(ctx, turnID); err != nil { s.logger.Warn("failed to complete running tool calls for turn", zap.String("turn_id", turnID), zap.Error(err)) } else if affected > 0 { s.logger.Info("completed stale running tool calls on turn end", @@ -71,7 +71,7 @@ func (s *Service) CompleteTurn(ctx context.Context, turnID string) error { } // Fetch the completed turn to get the completed_at timestamp - turn, err := s.repo.GetTurn(ctx, turnID) + turn, err := s.turns.GetTurn(ctx, turnID) if err != nil { s.logger.Warn("failed to refetch completed turn", zap.String("turn_id", turnID), zap.Error(err)) // Turn was likely deleted (task deletion race), skip publishing @@ -90,14 +90,14 @@ func (s *Service) CompleteTurn(ctx context.Context, turnID string) error { // GetActiveTurn returns the currently active (non-completed) turn for a session. func (s *Service) GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) { - return s.repo.GetActiveTurnBySessionID(ctx, sessionID) + return s.turns.GetActiveTurnBySessionID(ctx, sessionID) } // getOrStartTurn returns the active turn for a session, or starts a new one if none exists. // This is used to ensure messages always have a valid turn ID. func (s *Service) getOrStartTurn(ctx context.Context, sessionID string) (*models.Turn, error) { // First try to get an active turn - turn, err := s.repo.GetActiveTurnBySessionID(ctx, sessionID) + turn, err := s.turns.GetActiveTurnBySessionID(ctx, sessionID) if err == nil && turn != nil { return turn, nil } @@ -129,34 +129,34 @@ func (s *Service) publishTurnEvent(eventType string, turn *models.Turn) { // GetGitSnapshots retrieves git snapshots for a session func (s *Service) GetGitSnapshots(ctx context.Context, sessionID string, limit int) ([]*models.GitSnapshot, error) { - return s.repo.GetGitSnapshotsBySession(ctx, sessionID, limit) + return s.gitSnapshots.GetGitSnapshotsBySession(ctx, sessionID, limit) } // GetLatestGitSnapshot retrieves the latest git snapshot for a session func (s *Service) GetLatestGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) { - return s.repo.GetLatestGitSnapshot(ctx, sessionID) + return s.gitSnapshots.GetLatestGitSnapshot(ctx, sessionID) } // GetFirstGitSnapshot retrieves the first git snapshot for a session (oldest) func (s *Service) GetFirstGitSnapshot(ctx context.Context, sessionID string) (*models.GitSnapshot, error) { - return s.repo.GetFirstGitSnapshot(ctx, sessionID) + return s.gitSnapshots.GetFirstGitSnapshot(ctx, sessionID) } // GetSessionCommits retrieves commits for a session func (s *Service) GetSessionCommits(ctx context.Context, sessionID string) ([]*models.SessionCommit, error) { - return s.repo.GetSessionCommits(ctx, sessionID) + return s.gitSnapshots.GetSessionCommits(ctx, sessionID) } // GetLatestSessionCommit retrieves the latest commit for a session func (s *Service) GetLatestSessionCommit(ctx context.Context, sessionID string) (*models.SessionCommit, error) { - return s.repo.GetLatestSessionCommit(ctx, sessionID) + return s.gitSnapshots.GetLatestSessionCommit(ctx, sessionID) } // GetCumulativeDiff computes the cumulative diff from base commit to current HEAD // by using the first snapshot's base_commit and the latest snapshot's files func (s *Service) GetCumulativeDiff(ctx context.Context, sessionID string) (*models.CumulativeDiff, error) { // Get the first snapshot to find the base commit - firstSnapshot, err := s.repo.GetFirstGitSnapshot(ctx, sessionID) + firstSnapshot, err := s.gitSnapshots.GetFirstGitSnapshot(ctx, sessionID) if err != nil { if errors.Is(err, sql.ErrNoRows) { return nil, nil // No snapshots yet — valid state for fresh tasks @@ -165,13 +165,13 @@ func (s *Service) GetCumulativeDiff(ctx context.Context, sessionID string) (*mod } // Get the latest snapshot for current state - latestSnapshot, err := s.repo.GetLatestGitSnapshot(ctx, sessionID) + latestSnapshot, err := s.gitSnapshots.GetLatestGitSnapshot(ctx, sessionID) if err != nil { return nil, fmt.Errorf("failed to get latest git snapshot: %w", err) } // Count total commits for this session - commits, err := s.repo.GetSessionCommits(ctx, sessionID) + commits, err := s.gitSnapshots.GetSessionCommits(ctx, sessionID) if err != nil { return nil, fmt.Errorf("failed to get session commits: %w", err) } @@ -189,7 +189,7 @@ func (s *Service) GetCumulativeDiff(ctx context.Context, sessionID string) (*mod // This implements the lifecycle.WorkspaceInfoProvider interface. // The taskID parameter is optional - if empty, it will be looked up from the session. func (s *Service) GetWorkspaceInfoForSession(ctx context.Context, taskID, sessionID string) (*lifecycle.WorkspaceInfo, error) { - session, err := s.repo.GetTaskSession(ctx, sessionID) + session, err := s.sessions.GetTaskSession(ctx, sessionID) if err != nil { return nil, fmt.Errorf("failed to get session %s: %w", sessionID, err) } diff --git a/apps/backend/internal/task/service/service_workflow.go b/apps/backend/internal/task/service/service_workflow.go index 6f09100939..9d32b33e4d 100644 --- a/apps/backend/internal/task/service/service_workflow.go +++ b/apps/backend/internal/task/service/service_workflow.go @@ -26,14 +26,14 @@ type ApproveSessionResult struct { // If no transition actions are configured, it falls back to the next step by position. func (s *Service) ApproveSession(ctx context.Context, sessionID string) (*ApproveSessionResult, error) { // Update review status to approved - if err := s.repo.UpdateSessionReviewStatus(ctx, sessionID, "approved"); err != nil { + if err := s.sessions.UpdateSessionReviewStatus(ctx, sessionID, "approved"); err != nil { return nil, fmt.Errorf("failed to update review status: %w", err) } result := &ApproveSessionResult{} // Reload session to get updated review status - session, err := s.repo.GetTaskSession(ctx, sessionID) + session, err := s.sessions.GetTaskSession(ctx, sessionID) if err != nil { return nil, fmt.Errorf("failed to reload session: %w", err) } @@ -66,7 +66,7 @@ func (s *Service) applyApprovalStepTransition(ctx context.Context, sessionID str return } - if err := s.repo.UpdateSessionWorkflowStep(ctx, sessionID, newStepID); err != nil { + if err := s.sessions.UpdateSessionWorkflowStep(ctx, sessionID, newStepID); err != nil { s.logger.Error("failed to move session to next step after approval", zap.String("session_id", sessionID), zap.String("step_id", newStepID), @@ -75,14 +75,14 @@ func (s *Service) applyApprovalStepTransition(ctx context.Context, sessionID str } // Also move the task to the new step - if task, err := s.repo.GetTask(ctx, result.Session.TaskID); err != nil { + if task, err := s.tasks.GetTask(ctx, result.Session.TaskID); err != nil { s.logger.Error("failed to get task for approval transition", zap.String("task_id", result.Session.TaskID), zap.Error(err)) } else { task.WorkflowStepID = newStepID task.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateTask(ctx, task); err != nil { + if err := s.tasks.UpdateTask(ctx, task); err != nil { s.logger.Error("failed to move task to next step after approval", zap.String("task_id", result.Session.TaskID), zap.String("step_id", newStepID), @@ -94,7 +94,7 @@ func (s *Service) applyApprovalStepTransition(ctx context.Context, sessionID str } // Reload session with new step - result.Session, _ = s.repo.GetTaskSession(ctx, sessionID) + result.Session, _ = s.sessions.GetTaskSession(ctx, sessionID) // Get the new workflow step for the response if newStep, err := s.workflowStepGetter.GetStep(ctx, newStepID); err == nil { @@ -155,20 +155,20 @@ func (s *Service) resolveApprovalNextStep(ctx context.Context, step *wfmodels.Wo // UpdateTaskState updates the state of a task, moves it to the matching column, // and publishes a task.state_changed event func (s *Service) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) (*models.Task, error) { - task, err := s.repo.GetTask(ctx, id) + task, err := s.tasks.GetTask(ctx, id) if err != nil { return nil, err } oldState := task.State - if err := s.repo.UpdateTaskState(ctx, id, state); err != nil { + if err := s.tasks.UpdateTaskState(ctx, id, state); err != nil { s.logger.Error("failed to update task state", zap.String("task_id", id), zap.Error(err)) return nil, err } // Reload task to get updated state - task, err = s.repo.GetTask(ctx, id) + task, err = s.tasks.GetTask(ctx, id) if err != nil { return nil, err } @@ -189,7 +189,7 @@ func (s *Service) UpdateTaskState(ctx context.Context, id string, state v1.TaskS // UpdateTaskMetadata updates only the metadata of a task (merges with existing) func (s *Service) UpdateTaskMetadata(ctx context.Context, id string, metadata map[string]interface{}) (*models.Task, error) { - task, err := s.repo.GetTask(ctx, id) + task, err := s.tasks.GetTask(ctx, id) if err != nil { return nil, err } @@ -203,7 +203,7 @@ func (s *Service) UpdateTaskMetadata(ctx context.Context, id string, metadata ma } task.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateTask(ctx, task); err != nil { + if err := s.tasks.UpdateTask(ctx, task); err != nil { s.logger.Error("failed to update task metadata", zap.String("task_id", id), zap.Error(err)) return nil, err } @@ -220,7 +220,7 @@ type MoveTaskResult struct { // MoveTask moves a task to a different workflow step and position func (s *Service) MoveTask(ctx context.Context, id string, workflowID string, workflowStepID string, position int) (*MoveTaskResult, error) { - task, err := s.repo.GetTask(ctx, id) + task, err := s.tasks.GetTask(ctx, id) if err != nil { return nil, err } @@ -235,7 +235,7 @@ func (s *Service) MoveTask(ctx context.Context, id string, workflowID string, wo task.Position = position task.UpdatedAt = time.Now().UTC() - if err := s.repo.UpdateTask(ctx, task); err != nil { + if err := s.tasks.UpdateTask(ctx, task); err != nil { s.logger.Error("failed to move task", zap.String("task_id", id), zap.Error(err)) return nil, err } @@ -279,7 +279,7 @@ func (s *Service) checkMoveTaskApproval(ctx context.Context, taskID, currentStep if currentStepID == targetStepID { return nil } - primarySession, err := s.repo.GetPrimarySessionByTaskID(ctx, taskID) + primarySession, err := s.sessions.GetPrimarySessionByTaskID(ctx, taskID) if err != nil || primarySession == nil { return nil } @@ -295,14 +295,14 @@ func (s *Service) syncActiveSessionWorkflowStep(ctx context.Context, taskID, wor if workflowStepID == "" { return } - activeSession, err := s.repo.GetActiveTaskSessionByTaskID(ctx, taskID) + activeSession, err := s.sessions.GetActiveTaskSessionByTaskID(ctx, taskID) if err != nil || activeSession == nil { return } if activeSession.WorkflowStepID != nil && *activeSession.WorkflowStepID == workflowStepID { return } - if err := s.repo.UpdateSessionWorkflowStep(ctx, activeSession.ID, workflowStepID); err != nil { + if err := s.sessions.UpdateSessionWorkflowStep(ctx, activeSession.ID, workflowStepID); err != nil { s.logger.Warn("failed to update session workflow step after task move", zap.String("task_id", taskID), zap.String("session_id", activeSession.ID), @@ -318,12 +318,12 @@ func (s *Service) syncActiveSessionWorkflowStep(ctx context.Context, taskID, wor // CountTasksByWorkflow returns the number of tasks in a workflow func (s *Service) CountTasksByWorkflow(ctx context.Context, workflowID string) (int, error) { - return s.repo.CountTasksByWorkflow(ctx, workflowID) + return s.tasks.CountTasksByWorkflow(ctx, workflowID) } // CountTasksByWorkflowStep returns the number of tasks in a workflow step func (s *Service) CountTasksByWorkflowStep(ctx context.Context, stepID string) (int, error) { - return s.repo.CountTasksByWorkflowStep(ctx, stepID) + return s.tasks.CountTasksByWorkflowStep(ctx, stepID) } // BulkMoveTasksResult contains the result of a BulkMoveTasks operation. @@ -338,9 +338,9 @@ func (s *Service) BulkMoveTasks(ctx context.Context, sourceWorkflowID, sourceSte var tasks []*models.Task var err error if sourceStepID != "" { - tasks, err = s.repo.ListTasksByWorkflowStep(ctx, sourceStepID) + tasks, err = s.tasks.ListTasksByWorkflowStep(ctx, sourceStepID) } else { - tasks, err = s.repo.ListTasks(ctx, sourceWorkflowID) + tasks, err = s.tasks.ListTasks(ctx, sourceWorkflowID) } if err != nil { return nil, fmt.Errorf("failed to list tasks for bulk move: %w", err) @@ -357,7 +357,7 @@ func (s *Service) BulkMoveTasks(ctx context.Context, sourceWorkflowID, sourceSte task.Position = i task.UpdatedAt = now - if err := s.repo.UpdateTask(ctx, task); err != nil { + if err := s.tasks.UpdateTask(ctx, task); err != nil { s.logger.Error("failed to move task in bulk move", zap.String("task_id", task.ID), zap.Error(err)) @@ -365,10 +365,10 @@ func (s *Service) BulkMoveTasks(ctx context.Context, sourceWorkflowID, sourceSte } // Update active session's workflow_step_id - activeSession, err := s.repo.GetActiveTaskSessionByTaskID(ctx, task.ID) + activeSession, err := s.sessions.GetActiveTaskSessionByTaskID(ctx, task.ID) if err == nil && activeSession != nil { if activeSession.WorkflowStepID == nil || *activeSession.WorkflowStepID != targetStepID { - if err := s.repo.UpdateSessionWorkflowStep(ctx, activeSession.ID, targetStepID); err != nil { + if err := s.sessions.UpdateSessionWorkflowStep(ctx, activeSession.ID, targetStepID); err != nil { s.logger.Warn("failed to update session workflow step during bulk move", zap.String("task_id", task.ID), zap.String("session_id", activeSession.ID),