fix(mcp): allow parent workspace-source recovery (#2842)
Co-authored-by: ayattara <alassane.yattara@savoirfairelinux.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
@@ -1,12 +1,20 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/events/bus"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
taskrepository "github.com/kandev/kandev/internal/task/repository"
|
||||
sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
|
||||
"github.com/kandev/kandev/internal/task/service"
|
||||
ws "github.com/kandev/kandev/pkg/websocket"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
)
|
||||
@@ -35,3 +43,406 @@ func TestRegisterHandlers_LogsMeasuredRegistrationDelta(t *testing.T) {
|
||||
t.Fatalf("logged handler delta = %d, measured dispatcher delta = %d", count, delta)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesDirectParentAuthorizesAndAudits(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc, repo, parent, child := newWorkspaceSourceAuthorizationFixture(t)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "parent-session", TaskID: parent.ID}))
|
||||
|
||||
core, observed := observer.New(zap.InfoLevel)
|
||||
log, err := logger.NewFromZap(zap.New(core))
|
||||
require.NoError(t, err)
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: log.WithFields()}
|
||||
folder := t.TempDir()
|
||||
|
||||
response, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": parent.ID, "caller_session_id": "parent-session",
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": folder, "display_name": "docs"}},
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, ws.MessageTypeResponse, response.Type)
|
||||
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, child.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, folders, 1)
|
||||
require.Equal(t, "docs", folders[0].DisplayName)
|
||||
|
||||
entries := observed.FilterMessage("workspace sources attached through task MCP").All()
|
||||
require.Len(t, entries, 1)
|
||||
fields := entries[0].ContextMap()
|
||||
require.Equal(t, parent.ID, fields["caller_task_id"])
|
||||
require.Equal(t, "parent-session", fields["caller_session_id"])
|
||||
require.Equal(t, child.ID, fields["target_task_id"])
|
||||
require.Equal(t, int64(1), fields["requested_source_count"])
|
||||
require.Equal(t, true, fields["durable_state_changed"])
|
||||
require.NotContains(t, fmt.Sprint(fields), folder)
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesRejectsUnrelatedAndMismatchedCallers(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc, repo, parent, child := newWorkspaceSourceAuthorizationFixture(t)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "parent-session", TaskID: parent.ID}))
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "child-session", TaskID: child.ID}))
|
||||
sibling := createWorkspaceSourceAuthorizationTask(t, svc, parent.WorkspaceID, "Sibling", "")
|
||||
grandparent := createWorkspaceSourceAuthorizationTask(t, svc, parent.WorkspaceID, "Grandparent", "")
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "grandparent-session", TaskID: grandparent.ID}))
|
||||
_, err := repo.DB().ExecContext(ctx, "UPDATE tasks SET parent_id = ? WHERE id = ?", grandparent.ID, parent.ID)
|
||||
require.NoError(t, err)
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: testLogger(t).WithFields()}
|
||||
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
targetTaskID string
|
||||
callerSessionID string
|
||||
}{
|
||||
{name: "sibling", targetTaskID: sibling.ID, callerSessionID: "parent-session"},
|
||||
{name: "grandparent", targetTaskID: child.ID, callerSessionID: "grandparent-session"},
|
||||
{name: "mismatched session", targetTaskID: child.ID, callerSessionID: "child-session"},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
response, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": testCase.targetTaskID, "caller_task_id": parent.ID, "caller_session_id": testCase.callerSessionID,
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": t.TempDir(), "display_name": "docs"}},
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
assertWSError(t, response, ws.ErrorCodeForbidden)
|
||||
|
||||
folders, listErr := repo.ListTaskWorkspaceFolders(ctx, testCase.targetTaskID)
|
||||
require.NoError(t, listErr)
|
||||
require.Empty(t, folders)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesReturnsInternalErrorWhenCallerSessionLookupFails(t *testing.T) {
|
||||
h := &Handlers{
|
||||
taskSvc: &service.Service{},
|
||||
sessionRepo: failingCallerSessionRepository{err: errors.New("database unavailable")},
|
||||
}
|
||||
|
||||
response, err := h.handleAddWorkspaceSources(context.Background(), makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": "target-task", "caller_task_id": "caller-task", "caller_session_id": "caller-session",
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": t.TempDir(), "display_name": "docs"}},
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
assertWSError(t, response, ws.ErrorCodeInternalError)
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesReturnsInternalErrorWhenCallerTaskLookupFails(t *testing.T) {
|
||||
h := &Handlers{
|
||||
taskSvc: service.NewService(service.Repos{
|
||||
Tasks: failingCallerTaskRepository{err: errors.New("database unavailable")},
|
||||
}, nil, testLogger(t), service.RepositoryDiscoveryConfig{}),
|
||||
sessionRepo: staticCallerSessionRepository{session: &models.TaskSession{ID: "caller-session", TaskID: "caller-task"}},
|
||||
}
|
||||
|
||||
response, err := h.handleAddWorkspaceSources(context.Background(), makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": "target-task", "caller_task_id": "caller-task", "caller_session_id": "caller-session",
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": t.TempDir(), "display_name": "docs"}},
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
assertWSError(t, response, ws.ErrorCodeInternalError)
|
||||
}
|
||||
|
||||
type failingCallerSessionRepository struct{ err error }
|
||||
|
||||
func (r failingCallerSessionRepository) GetTaskSession(context.Context, string) (*models.TaskSession, error) {
|
||||
return nil, r.err
|
||||
}
|
||||
|
||||
func (failingCallerSessionRepository) UpdateTaskSessionState(context.Context, string, models.TaskSessionState, string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (failingCallerSessionRepository) SetSessionMetadataKeyIfAbsentOrDifferentStep(context.Context, string, string, string, interface{}) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
type staticCallerSessionRepository struct{ session *models.TaskSession }
|
||||
|
||||
func (r staticCallerSessionRepository) GetTaskSession(context.Context, string) (*models.TaskSession, error) {
|
||||
return r.session, nil
|
||||
}
|
||||
|
||||
func (staticCallerSessionRepository) UpdateTaskSessionState(context.Context, string, models.TaskSessionState, string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (staticCallerSessionRepository) SetSessionMetadataKeyIfAbsentOrDifferentStep(context.Context, string, string, string, interface{}) (bool, error) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
type failingCallerTaskRepository struct {
|
||||
taskrepository.TaskRepository
|
||||
err error
|
||||
}
|
||||
|
||||
func (r failingCallerTaskRepository) GetTask(context.Context, string) (*models.Task, error) {
|
||||
return nil, r.err
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesSelfTargetSucceeds(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc, repo, _, child, _ := newWorkspaceSourceAuthorizationFixtureWithEventBus(t)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "child-session", TaskID: child.ID}))
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: testLogger(t).WithFields()}
|
||||
|
||||
response, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": child.ID, "caller_session_id": "child-session",
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": t.TempDir(), "display_name": "docs"}},
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, ws.MessageTypeResponse, response.Type)
|
||||
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, child.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, folders, 1)
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesDenialsLeaveWorkspaceSourcesUntouched(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc, repo, parent, child, eventBus := newWorkspaceSourceAuthorizationFixtureWithEventBus(t)
|
||||
events := observeWorkspaceSourceEvents(t, eventBus)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "parent-session", TaskID: parent.ID}))
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "child-session", TaskID: child.ID}))
|
||||
materializer := &workspaceSourceMaterializerRecorder{}
|
||||
refresher := &workspaceSourceProviderRefresherRecorder{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: testLogger(t).WithFields()}
|
||||
|
||||
otherWorkspace := &models.Workspace{ID: "workspace-sources-other", Name: "Other workspace"}
|
||||
require.NoError(t, repo.CreateWorkspace(ctx, otherWorkspace))
|
||||
require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "workflow-sources-other", WorkspaceID: otherWorkspace.ID, Name: "Workflow"}))
|
||||
require.NoError(t, repo.CreateRepository(ctx, &models.Repository{ID: "repository-sources-other", WorkspaceID: otherWorkspace.ID, Name: "other", DefaultBranch: "main"}))
|
||||
otherParentResult, err := svc.CreateTask(ctx, &service.CreateTaskRequest{
|
||||
WorkspaceID: otherWorkspace.ID, WorkflowID: "workflow-sources-other", Title: "Other parent",
|
||||
Repositories: []service.TaskRepositoryInput{{RepositoryID: "repository-sources-other", BaseBranch: "main"}},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "other-parent-session", TaskID: otherParentResult.Task.ID}))
|
||||
_, err = repo.DB().ExecContext(ctx, "UPDATE tasks SET parent_id = ? WHERE id = ?", otherParentResult.Task.ID, child.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
callerTaskID string
|
||||
callerSessionID string
|
||||
expectedCode string
|
||||
}{
|
||||
{name: "cross workspace direct parent", callerTaskID: otherParentResult.Task.ID, callerSessionID: "other-parent-session", expectedCode: ws.ErrorCodeForbidden},
|
||||
{name: "unknown caller session", callerTaskID: parent.ID, callerSessionID: "missing-parent-session", expectedCode: ws.ErrorCodeForbidden},
|
||||
{name: "mismatched caller session", callerTaskID: parent.ID, callerSessionID: "child-session", expectedCode: ws.ErrorCodeForbidden},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
events.reset()
|
||||
materializer.called = false
|
||||
refresher.calls = 0
|
||||
response, handlerErr := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": testCase.callerTaskID, "caller_session_id": testCase.callerSessionID,
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": t.TempDir(), "display_name": "docs"}},
|
||||
}))
|
||||
require.NoError(t, handlerErr)
|
||||
assertWSError(t, response, testCase.expectedCode)
|
||||
assertWorkspaceSourceStateUnchanged(t, ctx, repo, child.ID, events, materializer, refresher, 1, 0)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesActiveChildRejectsExactRetryBeforeSideEffects(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc, repo, parent, child, eventBus := newWorkspaceSourceAuthorizationFixtureWithEventBus(t)
|
||||
events := observeWorkspaceSourceEvents(t, eventBus)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "parent-session", TaskID: parent.ID}))
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "child-session", TaskID: child.ID}))
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: testLogger(t).WithFields()}
|
||||
folder := t.TempDir()
|
||||
source := []interface{}{map[string]interface{}{"kind": "folder", "local_path": folder, "display_name": "docs"}}
|
||||
|
||||
initial, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": parent.ID, "caller_session_id": "parent-session", "sources": source,
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, ws.MessageTypeResponse, initial.Type)
|
||||
_, err = svc.StartTurn(ctx, "child-session")
|
||||
require.NoError(t, err)
|
||||
|
||||
materializer := &workspaceSourceMaterializerRecorder{}
|
||||
refresher := &workspaceSourceProviderRefresherRecorder{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
events.reset()
|
||||
retry, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": parent.ID, "caller_session_id": "parent-session", "sources": source,
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
assertWSError(t, retry, ws.ErrorCodeConflict)
|
||||
assertWorkspaceSourceStateUnchanged(t, ctx, repo, child.ID, events, materializer, refresher, 1, 1)
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesRejectsChildReparentedBeforeDurableCommit(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc, repo, parent, child, eventBus, _ := newWorkspaceSourceAuthorizationFixtureWithReparentingStore(t)
|
||||
events := observeWorkspaceSourceEvents(t, eventBus)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "parent-session", TaskID: parent.ID}))
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "child-session", TaskID: child.ID}))
|
||||
materializer := &workspaceSourceMaterializerRecorder{}
|
||||
refresher := &workspaceSourceProviderRefresherRecorder{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: testLogger(t).WithFields()}
|
||||
|
||||
response, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": parent.ID, "caller_session_id": "parent-session",
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": t.TempDir(), "display_name": "docs"}},
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
assertWSError(t, response, ws.ErrorCodeForbidden)
|
||||
assertWorkspaceSourceStateUnchanged(t, ctx, repo, child.ID, events, materializer, refresher, 1, 0)
|
||||
|
||||
session, err := repo.GetTaskSession(ctx, "child-session")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, models.TaskSessionStateCreated, session.State)
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesRejectsExactRetryAfterChildReparenting(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
svc, repo, parent, child, eventBus, reparentingStore := newWorkspaceSourceAuthorizationFixtureWithReparentingStore(t)
|
||||
events := observeWorkspaceSourceEvents(t, eventBus)
|
||||
reparentingStore.beforeCreate = nil
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "parent-session", TaskID: parent.ID}))
|
||||
materializer := &workspaceSourceMaterializerRecorder{}
|
||||
refresher := &workspaceSourceProviderRefresherRecorder{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: testLogger(t).WithFields()}
|
||||
source := []interface{}{map[string]interface{}{"kind": "folder", "local_path": t.TempDir(), "display_name": "docs"}}
|
||||
|
||||
initial, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": parent.ID, "caller_session_id": "parent-session", "sources": source,
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, ws.MessageTypeResponse, initial.Type)
|
||||
events.reset()
|
||||
materializer.called, refresher.calls = false, 0
|
||||
reparentingStore.beforeCreate = func(context.Context) error {
|
||||
_, err := repo.DB().ExecContext(ctx, "UPDATE tasks SET parent_id = '' WHERE id = ?", child.ID)
|
||||
return err
|
||||
}
|
||||
|
||||
retry, err := h.handleAddWorkspaceSources(ctx, makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": child.ID, "caller_task_id": parent.ID, "caller_session_id": "parent-session", "sources": source,
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
assertWSError(t, retry, ws.ErrorCodeForbidden)
|
||||
assertWorkspaceSourceStateUnchanged(t, ctx, repo, child.ID, events, materializer, refresher, 1, 1)
|
||||
}
|
||||
|
||||
type workspaceSourceMaterializerRecorder struct{ called bool }
|
||||
|
||||
func (m *workspaceSourceMaterializerRecorder) MaterializeWorkspaceSources(context.Context, string, *models.WorkspaceSourceBatch) (*service.WorkspaceSourceMaterializationResult, error) {
|
||||
m.called = true
|
||||
return &service.WorkspaceSourceMaterializationResult{}, nil
|
||||
}
|
||||
|
||||
type workspaceSourceProviderRefresherRecorder struct{ calls int }
|
||||
|
||||
func (r *workspaceSourceProviderRefresherRecorder) RefreshTaskMCPProviders(context.Context, string) error {
|
||||
r.calls++
|
||||
return nil
|
||||
}
|
||||
|
||||
type workspaceSourceEventObserver struct{ count int }
|
||||
|
||||
func observeWorkspaceSourceEvents(t *testing.T, eventBus *bus.MemoryEventBus) *workspaceSourceEventObserver {
|
||||
t.Helper()
|
||||
observer := &workspaceSourceEventObserver{}
|
||||
subscription, err := eventBus.Subscribe(">", func(context.Context, *bus.Event) error {
|
||||
observer.count++
|
||||
return nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = subscription.Unsubscribe() })
|
||||
return observer
|
||||
}
|
||||
|
||||
func (o *workspaceSourceEventObserver) reset() { o.count = 0 }
|
||||
|
||||
func assertWorkspaceSourceStateUnchanged(t *testing.T, ctx context.Context, repo *sqliterepo.Repository, taskID string, events *workspaceSourceEventObserver, materializer *workspaceSourceMaterializerRecorder, refresher *workspaceSourceProviderRefresherRecorder, wantRepositories, wantFolders int) {
|
||||
t.Helper()
|
||||
repositories, err := repo.ListTaskRepositories(ctx, taskID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, repositories, wantRepositories)
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, taskID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, folders, wantFolders)
|
||||
require.Zero(t, events.count)
|
||||
require.False(t, materializer.called)
|
||||
require.Zero(t, refresher.calls)
|
||||
}
|
||||
|
||||
func newWorkspaceSourceAuthorizationFixture(t *testing.T) (*service.Service, *sqliterepo.Repository, *models.Task, *models.Task) {
|
||||
svc, repo, parent, child, _ := newWorkspaceSourceAuthorizationFixtureWithEventBus(t)
|
||||
return svc, repo, parent, child
|
||||
}
|
||||
|
||||
func newWorkspaceSourceAuthorizationFixtureWithEventBus(t *testing.T) (*service.Service, *sqliterepo.Repository, *models.Task, *models.Task, *bus.MemoryEventBus) {
|
||||
t.Helper()
|
||||
svc, repo, eventBus := newTestTaskServiceWithEventBus(t)
|
||||
ctx := context.Background()
|
||||
require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "workspace-sources", Name: "Workspace sources"}))
|
||||
require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "workflow-sources", WorkspaceID: "workspace-sources", Name: "Workflow"}))
|
||||
require.NoError(t, repo.CreateRepository(ctx, &models.Repository{ID: "repository-sources", WorkspaceID: "workspace-sources", Name: "app", DefaultBranch: "main"}))
|
||||
parent := createWorkspaceSourceAuthorizationTask(t, svc, "workspace-sources", "Parent", "")
|
||||
child := createWorkspaceSourceAuthorizationTask(t, svc, "workspace-sources", "Child", parent.ID)
|
||||
return svc, repo, parent, child, eventBus
|
||||
}
|
||||
|
||||
func newWorkspaceSourceAuthorizationFixtureWithReparentingStore(t *testing.T) (*service.Service, *sqliterepo.Repository, *models.Task, *models.Task, *bus.MemoryEventBus, *reparentingWorkspaceSourceStore) {
|
||||
t.Helper()
|
||||
_, repo, eventBus := newTestTaskServiceWithEventBus(t)
|
||||
ctx := context.Background()
|
||||
store := &reparentingWorkspaceSourceStore{TaskWorkspaceFolderRepository: repo}
|
||||
svc := service.NewService(service.Repos{
|
||||
Workspaces: repo, Tasks: repo, TaskRepos: repo, WorkspaceFolders: store, Workflows: repo,
|
||||
Messages: repo, Turns: repo, Sessions: repo, GitSnapshots: repo, RepoEntities: repo,
|
||||
Executors: repo, Environments: repo, Reviews: repo,
|
||||
}, eventBus, testLogger(t), service.RepositoryDiscoveryConfig{})
|
||||
require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "workspace-sources", Name: "Workspace sources"}))
|
||||
require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "workflow-sources", WorkspaceID: "workspace-sources", Name: "Workflow"}))
|
||||
require.NoError(t, repo.CreateRepository(ctx, &models.Repository{ID: "repository-sources", WorkspaceID: "workspace-sources", Name: "app", DefaultBranch: "main"}))
|
||||
parent := createWorkspaceSourceAuthorizationTask(t, svc, "workspace-sources", "Parent", "")
|
||||
child := createWorkspaceSourceAuthorizationTask(t, svc, "workspace-sources", "Child", parent.ID)
|
||||
store.beforeCreate = func(context.Context) error {
|
||||
_, err := repo.DB().ExecContext(ctx, "UPDATE tasks SET parent_id = '' WHERE id = ?", child.ID)
|
||||
return err
|
||||
}
|
||||
return svc, repo, parent, child, eventBus, store
|
||||
}
|
||||
|
||||
type reparentingWorkspaceSourceStore struct {
|
||||
taskrepository.TaskWorkspaceFolderRepository
|
||||
beforeCreate func(context.Context) error
|
||||
}
|
||||
|
||||
func (s *reparentingWorkspaceSourceStore) CreateWorkspaceSourceBatch(ctx context.Context, batch *models.WorkspaceSourceBatch) error {
|
||||
if s.beforeCreate != nil {
|
||||
if err := s.beforeCreate(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return s.TaskWorkspaceFolderRepository.CreateWorkspaceSourceBatch(ctx, batch)
|
||||
}
|
||||
|
||||
func createWorkspaceSourceAuthorizationTask(t *testing.T, svc *service.Service, workspaceID, title, parentID string) *models.Task {
|
||||
t.Helper()
|
||||
result, err := svc.CreateTask(context.Background(), &service.CreateTaskRequest{
|
||||
WorkspaceID: workspaceID,
|
||||
WorkflowID: "workflow-sources",
|
||||
Title: title,
|
||||
ParentID: parentID,
|
||||
Repositories: []service.TaskRepositoryInput{{RepositoryID: "repository-sources", BaseBranch: "main"}},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
return result.Task
|
||||
}
|
||||
|
||||
@@ -1829,29 +1829,109 @@ func (h *Handlers) handleAddBranchToTask(ctx context.Context, msg *ws.Message) (
|
||||
// union to the shared attachment service. It deliberately rejects fields from
|
||||
// the other variant so callers cannot get a silently ambiguous attachment.
|
||||
func (h *Handlers) handleAddWorkspaceSources(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
|
||||
var req struct {
|
||||
TaskID string `json:"task_id"`
|
||||
Sources []json.RawMessage `json:"sources"`
|
||||
req, sources, response := parseAddWorkspaceSourcesRequest(msg)
|
||||
if response != nil {
|
||||
return response, nil
|
||||
}
|
||||
if err := json.Unmarshal(msg.Payload, &req); err != nil {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
|
||||
caller, response := h.verifyWorkspaceSourceCaller(ctx, msg, req)
|
||||
if response != nil {
|
||||
return response, nil
|
||||
}
|
||||
if req.TaskID == "" || len(req.Sources) == 0 {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id and at least one source are required", nil)
|
||||
_, isChildTarget, response := h.authorizeWorkspaceSourceTarget(ctx, msg, req, caller)
|
||||
if response != nil {
|
||||
return response, nil
|
||||
}
|
||||
sources, err := parseWorkspaceSources(req.Sources)
|
||||
if err != nil {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, err.Error(), nil)
|
||||
attachReq := service.AttachWorkspaceSourcesRequest{TaskID: req.TaskID, Sources: sources}
|
||||
if isChildTarget {
|
||||
attachReq.ExpectedParentID = caller.ID
|
||||
attachReq.ExpectedParentWorkspaceID = caller.WorkspaceID
|
||||
}
|
||||
result, err := h.taskSvc.AttachWorkspaceSources(ctx, service.AttachWorkspaceSourcesRequest{TaskID: req.TaskID, Sources: sources})
|
||||
result, err := h.taskSvc.AttachWorkspaceSources(ctx, attachReq)
|
||||
if err != nil {
|
||||
return ws.NewError(msg.ID, msg.Action, classifyWorkspaceSourceError(err), "Failed to attach workspace sources: "+err.Error(), nil)
|
||||
}
|
||||
if h.logger != nil {
|
||||
h.logger.Info("workspace sources attached through task MCP",
|
||||
zap.String("caller_task_id", req.CallerTaskID), zap.String("caller_session_id", req.CallerSessionID),
|
||||
zap.String("target_task_id", req.TaskID), zap.Int("requested_source_count", len(req.Sources)), zap.Bool("durable_state_changed", result.Changed))
|
||||
}
|
||||
return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
|
||||
"task_id": result.Task.ID, "repositories": result.Task.Repositories, "workspace_folders": result.Task.WorkspaceFolders, "workspace_path": result.WorkspacePath, "session_ids": result.SessionIDs,
|
||||
})
|
||||
}
|
||||
|
||||
type addWorkspaceSourcesRequest struct {
|
||||
TaskID string `json:"task_id"`
|
||||
CallerTaskID string `json:"caller_task_id"`
|
||||
CallerSessionID string `json:"caller_session_id"`
|
||||
Sources []json.RawMessage `json:"sources"`
|
||||
}
|
||||
|
||||
func parseAddWorkspaceSourcesRequest(msg *ws.Message) (addWorkspaceSourcesRequest, []service.WorkspaceSourceInput, *ws.Message) {
|
||||
var req addWorkspaceSourcesRequest
|
||||
if err := json.Unmarshal(msg.Payload, &req); err != nil {
|
||||
return req, nil, newWorkspaceSourceError(msg, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error())
|
||||
}
|
||||
if req.TaskID == "" || req.CallerTaskID == "" || req.CallerSessionID == "" || len(req.Sources) == 0 {
|
||||
return req, nil, newWorkspaceSourceError(msg, ws.ErrorCodeValidation, "task_id, caller provenance, and at least one source are required")
|
||||
}
|
||||
sources, err := parseWorkspaceSources(req.Sources)
|
||||
if err != nil {
|
||||
return req, nil, newWorkspaceSourceError(msg, ws.ErrorCodeValidation, err.Error())
|
||||
}
|
||||
return req, sources, nil
|
||||
}
|
||||
|
||||
func (h *Handlers) verifyWorkspaceSourceCaller(ctx context.Context, msg *ws.Message, req addWorkspaceSourcesRequest) (*models.Task, *ws.Message) {
|
||||
if h.sessionRepo == nil || h.taskSvc == nil {
|
||||
return nil, newWorkspaceSourceError(msg, ws.ErrorCodeInternalError, "workspace source authorization is not configured")
|
||||
}
|
||||
callerSession, err := h.sessionRepo.GetTaskSession(ctx, req.CallerSessionID)
|
||||
if err != nil {
|
||||
if errors.Is(err, models.ErrTaskSessionNotFound) {
|
||||
return nil, newWorkspaceSourceError(msg, ws.ErrorCodeForbidden, "caller session is not authorized for the calling task")
|
||||
}
|
||||
return nil, newWorkspaceSourceError(msg, ws.ErrorCodeInternalError, "failed to verify caller session")
|
||||
}
|
||||
if callerSession == nil || callerSession.TaskID != req.CallerTaskID {
|
||||
return nil, newWorkspaceSourceError(msg, ws.ErrorCodeForbidden, "caller session is not authorized for the calling task")
|
||||
}
|
||||
caller, err := h.taskSvc.GetTask(ctx, req.CallerTaskID)
|
||||
if err != nil {
|
||||
if errors.Is(err, taskrepository.ErrTaskNotFound) {
|
||||
return nil, newWorkspaceSourceError(msg, ws.ErrorCodeForbidden, "caller task is not authorized")
|
||||
}
|
||||
return nil, newWorkspaceSourceError(msg, ws.ErrorCodeInternalError, "failed to verify caller task")
|
||||
}
|
||||
if caller == nil {
|
||||
return nil, newWorkspaceSourceError(msg, ws.ErrorCodeForbidden, "caller task is not authorized")
|
||||
}
|
||||
return caller, nil
|
||||
}
|
||||
|
||||
func (h *Handlers) authorizeWorkspaceSourceTarget(ctx context.Context, msg *ws.Message, req addWorkspaceSourcesRequest, caller *models.Task) (*models.Task, bool, *ws.Message) {
|
||||
target, err := h.taskSvc.GetTask(ctx, req.TaskID)
|
||||
if err != nil {
|
||||
if errors.Is(err, taskrepository.ErrTaskNotFound) {
|
||||
return nil, false, newWorkspaceSourceError(msg, ws.ErrorCodeNotFound, "target task not found")
|
||||
}
|
||||
return nil, false, newWorkspaceSourceError(msg, ws.ErrorCodeInternalError, "failed to load target task")
|
||||
}
|
||||
if target == nil {
|
||||
return nil, false, newWorkspaceSourceError(msg, ws.ErrorCodeNotFound, "target task not found")
|
||||
}
|
||||
isChildTarget := req.TaskID != req.CallerTaskID
|
||||
if isChildTarget && !canDirectParentAccess(caller, target) {
|
||||
return nil, false, newWorkspaceSourceError(msg, ws.ErrorCodeForbidden, "only a task's direct parent in the same workspace can attach its sources")
|
||||
}
|
||||
return target, isChildTarget, nil
|
||||
}
|
||||
|
||||
func newWorkspaceSourceError(msg *ws.Message, code, message string) *ws.Message {
|
||||
response, _ := ws.NewError(msg.ID, msg.Action, code, message, nil)
|
||||
return response
|
||||
}
|
||||
|
||||
func parseWorkspaceSources(raw []json.RawMessage) ([]service.WorkspaceSourceInput, error) {
|
||||
sources := make([]service.WorkspaceSourceInput, 0, len(raw))
|
||||
for _, item := range raw {
|
||||
@@ -1890,6 +1970,8 @@ func parseWorkspaceSources(raw []json.RawMessage) ([]service.WorkspaceSourceInpu
|
||||
|
||||
func classifyWorkspaceSourceError(err error) string {
|
||||
switch {
|
||||
case errors.Is(err, taskrepository.ErrTaskParentMismatch):
|
||||
return ws.ErrorCodeForbidden
|
||||
case errors.Is(err, service.ErrInvalidWorkspaceSource):
|
||||
return ws.ErrorCodeValidation
|
||||
case errors.Is(err, taskrepo.ErrTaskNotFound), errors.Is(err, taskrepository.ErrRepositoryNotFound), errors.Is(err, service.ErrTaskRepositoryNotFound):
|
||||
|
||||
@@ -84,18 +84,19 @@ func newTestTaskServiceWithEventBus(t *testing.T) (*service.Service, *sqliterepo
|
||||
eventBus := bus.NewMemoryEventBus(log)
|
||||
t.Cleanup(func() { eventBus.Close() })
|
||||
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,
|
||||
Workspaces: repo,
|
||||
Tasks: repo,
|
||||
TaskRepos: repo,
|
||||
WorkspaceFolders: repo,
|
||||
Workflows: repo,
|
||||
Messages: repo,
|
||||
Turns: repo,
|
||||
Sessions: repo,
|
||||
GitSnapshots: repo,
|
||||
RepoEntities: repo,
|
||||
Executors: repo,
|
||||
Environments: repo,
|
||||
Reviews: repo,
|
||||
}, eventBus, log, service.RepositoryDiscoveryConfig{})
|
||||
return svc, repo, eventBus
|
||||
}
|
||||
@@ -122,18 +123,19 @@ func newTestTaskServiceWithWorkflow(t *testing.T) (*service.Service, *sqliterepo
|
||||
eventBus := bus.NewMemoryEventBus(log)
|
||||
t.Cleanup(func() { eventBus.Close() })
|
||||
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,
|
||||
Workspaces: repo,
|
||||
Tasks: repo,
|
||||
TaskRepos: repo,
|
||||
WorkspaceFolders: repo,
|
||||
Workflows: repo,
|
||||
Messages: repo,
|
||||
Turns: repo,
|
||||
Sessions: repo,
|
||||
GitSnapshots: repo,
|
||||
RepoEntities: repo,
|
||||
Executors: repo,
|
||||
Environments: repo,
|
||||
Reviews: repo,
|
||||
}, eventBus, log, service.RepositoryDiscoveryConfig{})
|
||||
workflowSvc := workflowservice.NewService(workflowRepo, log)
|
||||
t.Cleanup(func() { _ = workflowSvc.Close() })
|
||||
@@ -469,15 +471,24 @@ func TestHandleAddBranchToTask_ReturnsMaterializedPaths(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestHandleAddWorkspaceSourcesRejectsUnknownKind(t *testing.T) {
|
||||
h := &Handlers{}
|
||||
ctx := context.Background()
|
||||
svc, repo, parent, child := newWorkspaceSourceAuthorizationFixture(t)
|
||||
require.NoError(t, repo.CreateTaskSession(ctx, &models.TaskSession{ID: "parent-session", TaskID: parent.ID}))
|
||||
h := &Handlers{taskSvc: svc, sessionRepo: repo, logger: testLogger(t).WithFields()}
|
||||
msg := makeWSMessage(t, ws.ActionMCPAddWorkspaceSources, map[string]interface{}{
|
||||
"task_id": "task-1",
|
||||
"task_id": child.ID, "caller_task_id": parent.ID, "caller_session_id": "parent-session",
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "unknown"}},
|
||||
})
|
||||
|
||||
resp, err := h.handleAddWorkspaceSources(context.Background(), msg)
|
||||
resp, err := h.handleAddWorkspaceSources(ctx, msg)
|
||||
require.NoError(t, err)
|
||||
assertWSError(t, resp, ws.ErrorCodeValidation)
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, child.ID)
|
||||
require.NoError(t, err)
|
||||
require.Empty(t, folders)
|
||||
repositories, err := repo.ListTaskRepositories(ctx, child.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, repositories, 1)
|
||||
}
|
||||
|
||||
func TestClassifyWorkspaceSourceErrorMapsRepositoryNotFound(t *testing.T) {
|
||||
|
||||
@@ -108,9 +108,5 @@ func newStopTaskFailure(msg *ws.Message, code, message string) *stopTaskFailure
|
||||
}
|
||||
|
||||
func canStopTask(sender, target *models.Task) bool {
|
||||
return sender.WorkspaceID != "" &&
|
||||
target.WorkspaceID != "" &&
|
||||
target.ParentID != "" &&
|
||||
target.ParentID == sender.ID &&
|
||||
target.WorkspaceID == sender.WorkspaceID
|
||||
return canDirectParentAccess(sender, target)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
package handlers
|
||||
|
||||
import "github.com/kandev/kandev/internal/task/models"
|
||||
|
||||
// canDirectParentAccess is deliberately narrower than workspace membership:
|
||||
// only a direct parent may mutate its child through task-bound MCP.
|
||||
func canDirectParentAccess(caller, target *models.Task) bool {
|
||||
return caller != nil && target != nil && caller.WorkspaceID != "" &&
|
||||
target.WorkspaceID == caller.WorkspaceID && target.ParentID == caller.ID
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestCanDirectParentAccess(t *testing.T) {
|
||||
parent := &models.Task{ID: "parent", WorkspaceID: "workspace"}
|
||||
for _, tt := range []struct {
|
||||
name string
|
||||
target *models.Task
|
||||
want bool
|
||||
}{
|
||||
{name: "direct child", target: &models.Task{ParentID: "parent", WorkspaceID: "workspace"}, want: true},
|
||||
{name: "self", target: parent, want: false},
|
||||
{name: "sibling", target: &models.Task{ParentID: "other", WorkspaceID: "workspace"}, want: false},
|
||||
{name: "grandchild", target: &models.Task{ParentID: "child", WorkspaceID: "workspace"}, want: false},
|
||||
{name: "other workspace", target: &models.Task{ParentID: "parent", WorkspaceID: "other"}, want: false},
|
||||
{name: "empty workspace", target: &models.Task{ParentID: "parent"}, want: false},
|
||||
} {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
assert.Equal(t, tt.want, canDirectParentAccess(parent, tt.target))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -643,6 +643,9 @@ func TestAddBranchToTask_RejectsAnotherTask(t *testing.T) {
|
||||
func TestAddWorkspaceSourcesDefaultsTaskAndForwardsMixedSources(t *testing.T) {
|
||||
backend := &testBackend{response: map[string]interface{}{"task_id": "task-current"}}
|
||||
s := newTaskModeServer(t, backend, "task-current")
|
||||
properties := toolInputProperties(t, s, "add_workspace_sources_kandev")
|
||||
assert.NotContains(t, properties, "caller_task_id")
|
||||
assert.NotContains(t, properties, "caller_session_id")
|
||||
|
||||
result := callTool(t, s, "add_workspace_sources_kandev", map[string]interface{}{
|
||||
"sources": []interface{}{
|
||||
@@ -656,11 +659,13 @@ func TestAddWorkspaceSourcesDefaultsTaskAndForwardsMixedSources(t *testing.T) {
|
||||
payload, ok := backend.lastPayload.(map[string]interface{})
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "task-current", payload["task_id"])
|
||||
assert.Equal(t, "task-current", payload["caller_task_id"])
|
||||
assert.Equal(t, "test-session", payload["caller_session_id"])
|
||||
assert.Len(t, payload["sources"], 2)
|
||||
}
|
||||
|
||||
func TestAddWorkspaceSourcesRejectsAnotherTask(t *testing.T) {
|
||||
backend := &testBackend{}
|
||||
func TestAddWorkspaceSourcesForwardsDirectChildTargetWithTrustedProvenance(t *testing.T) {
|
||||
backend := &testBackend{response: map[string]interface{}{"task_id": "task-child"}}
|
||||
s := newTaskModeServer(t, backend, "task-current")
|
||||
|
||||
result := callTool(t, s, "add_workspace_sources_kandev", map[string]interface{}{
|
||||
@@ -668,8 +673,12 @@ func TestAddWorkspaceSourcesRejectsAnotherTask(t *testing.T) {
|
||||
"sources": []interface{}{map[string]interface{}{"kind": "folder", "local_path": "/tmp/docs"}},
|
||||
})
|
||||
|
||||
assert.True(t, result.IsError)
|
||||
assert.Empty(t, backend.lastAction)
|
||||
assert.False(t, result.IsError)
|
||||
payload, ok := backend.lastPayload.(map[string]interface{})
|
||||
require.True(t, ok)
|
||||
assert.Equal(t, "task-other", payload["task_id"])
|
||||
assert.Equal(t, "task-current", payload["caller_task_id"])
|
||||
assert.Equal(t, "test-session", payload["caller_session_id"])
|
||||
}
|
||||
|
||||
// TestAddBranchToTask_ForwardsLocalPath verifies local_path is plumbed through
|
||||
|
||||
@@ -1301,8 +1301,12 @@ func (s *Server) registerAddBranchToTaskTool() {
|
||||
func (s *Server) registerAddWorkspaceSourcesTool() {
|
||||
s.mcpServer.AddTool(
|
||||
mcp.NewTool("add_workspace_sources_kandev",
|
||||
mcp.WithDescription("Attach repository and folder workspace sources to an idle task. task_id defaults to the current task."),
|
||||
mcp.WithDescription("Attach repository and folder workspace sources to an idle task. A task may update itself or its same-workspace direct child. Exact retries are idempotent."),
|
||||
mcp.WithString(mcpKeyTaskID, mcp.Description("Task to update. Defaults to the current task.")),
|
||||
mcp.WithReadOnlyHintAnnotation(false),
|
||||
mcp.WithDestructiveHintAnnotation(false),
|
||||
mcp.WithIdempotentHintAnnotation(true),
|
||||
mcp.WithOpenWorldHintAnnotation(true),
|
||||
mcp.WithArray("sources", mcp.Required(), mcp.MinItems(1),
|
||||
mcp.Description("Ordered source objects. Each has kind repository or folder and the documented fields for that kind."),
|
||||
mcp.Items(map[string]any{"type": "object"}),
|
||||
@@ -1321,11 +1325,8 @@ func (s *Server) addWorkspaceSourcesHandler() server.ToolHandlerFunc {
|
||||
if taskID == "" {
|
||||
return mcp.NewToolResultError("task_id is required (no current task context to default to)"), nil
|
||||
}
|
||||
// A task-mode MCP server is bound to one live task. Never let an agent
|
||||
// use this mutation tool as a cross-task capability: the backend tunnel
|
||||
// has no authenticated user principal to authorize arbitrary task IDs.
|
||||
if s.taskID == "" || taskID != s.taskID {
|
||||
return mcp.NewToolResultError("task_id is not available in this session"), nil
|
||||
if s.taskID == "" || s.sessionID == "" {
|
||||
return mcp.NewToolResultError("task source provenance is unavailable in this session"), nil
|
||||
}
|
||||
arguments, ok := req.Params.Arguments.(map[string]interface{})
|
||||
if !ok {
|
||||
@@ -1335,7 +1336,8 @@ func (s *Server) addWorkspaceSourcesHandler() server.ToolHandlerFunc {
|
||||
if !ok {
|
||||
return mcp.NewToolResultError("sources is required"), nil
|
||||
}
|
||||
payload := map[string]interface{}{mcpKeyTaskID: taskID, "sources": sources}
|
||||
// Provenance is server-authored and intentionally absent from the callable schema.
|
||||
payload := map[string]interface{}{mcpKeyTaskID: taskID, "sources": sources, "caller_task_id": s.taskID, "caller_session_id": s.sessionID}
|
||||
var result map[string]interface{}
|
||||
if err := s.backend.RequestPayload(ctx, ws.ActionMCPAddWorkspaceSources, payload, &result); err != nil {
|
||||
return mcp.NewToolResultError(err.Error()), nil
|
||||
|
||||
@@ -19,6 +19,7 @@ func TestCoreToolDescriptionsStayFocused(t *testing.T) {
|
||||
"stop_task_kandev": 700,
|
||||
"spawn_session_kandev": 600,
|
||||
"add_branch_to_task_kandev": 650,
|
||||
"add_workspace_sources_kandev": 500,
|
||||
"step_complete_kandev": 650,
|
||||
"ask_user_question_kandev": 600,
|
||||
"show_rich_output_kandev": 650,
|
||||
@@ -68,6 +69,7 @@ func TestCoreToolRiskAnnotations(t *testing.T) {
|
||||
{name: "archive_task_kandev", readOnly: false, destructive: false, idempotent: true, openWorld: false},
|
||||
{name: "delete_task_kandev", readOnly: false, destructive: true, idempotent: false, openWorld: false},
|
||||
{name: "stop_task_kandev", readOnly: false, destructive: true, idempotent: true, openWorld: false},
|
||||
{name: "add_workspace_sources_kandev", readOnly: false, destructive: false, idempotent: true, openWorld: true},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
|
||||
@@ -1035,9 +1035,11 @@ type TaskWorkspaceFolder struct {
|
||||
// attachment operation, so a later materialization failure can compensate
|
||||
// without touching pre-existing sources.
|
||||
type WorkspaceSourceBatch struct {
|
||||
TaskID string `json:"task_id"`
|
||||
Sources []WorkspaceSource `json:"sources,omitempty"`
|
||||
RepositoryUpdates []WorkspaceSourceRepositoryUpdate `json:"repository_updates,omitempty"`
|
||||
TaskID string `json:"task_id"`
|
||||
Sources []WorkspaceSource `json:"sources,omitempty"`
|
||||
RepositoryUpdates []WorkspaceSourceRepositoryUpdate `json:"repository_updates,omitempty"`
|
||||
ExpectedParentID string `json:"-"`
|
||||
ExpectedParentWorkspaceID string `json:"-"`
|
||||
}
|
||||
|
||||
// WorkspaceSourceRepositoryUpdate records a legacy association branch derived
|
||||
|
||||
@@ -15,6 +15,7 @@ import (
|
||||
var ErrWorkspaceNameMismatch = repoerrors.ErrWorkspaceNameMismatch
|
||||
var ErrWorkspaceNotFound = repoerrors.ErrWorkspaceNotFound
|
||||
var ErrTaskNotFound = repoerrors.ErrTaskNotFound
|
||||
var ErrTaskParentMismatch = repoerrors.ErrTaskParentMismatch
|
||||
var ErrTaskPlanNotFound = repoerrors.ErrTaskPlanNotFound
|
||||
var ErrRepositoryNotFound = repoerrors.ErrRepositoryNotFound
|
||||
var ErrTaskEnvironmentNotFound = repoerrors.ErrTaskEnvironmentNotFound
|
||||
|
||||
@@ -12,6 +12,10 @@ var ErrWorkspaceNotFound = errors.New("workspace not found")
|
||||
// ErrTaskNotFound reports that no task row matched the supplied id.
|
||||
var ErrTaskNotFound = errors.New("task not found")
|
||||
|
||||
// ErrTaskParentMismatch reports that a task no longer has the parent/workspace
|
||||
// relation a cross-task mutation was authorized against.
|
||||
var ErrTaskParentMismatch = errors.New("task parent relation no longer matches")
|
||||
|
||||
// ErrTaskPlanNotFound reports that no task plan row matched the supplied task id.
|
||||
var ErrTaskPlanNotFound = errors.New("task plan not found")
|
||||
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/jmoiron/sqlx"
|
||||
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
"github.com/kandev/kandev/internal/task/repository/repoerrors"
|
||||
)
|
||||
|
||||
// ListTaskWorkspaceFoldersByTaskIDs loads folders for a task batch in one query.
|
||||
@@ -89,6 +90,9 @@ func (r *Repository) CreateWorkspaceSourceBatch(ctx context.Context, batch *mode
|
||||
}
|
||||
|
||||
func (r *Repository) createWorkspaceSourceBatchTx(ctx context.Context, tx *sqlx.Tx, batch *models.WorkspaceSourceBatch) error {
|
||||
if err := r.guardWorkspaceSourceParentTx(ctx, tx, batch); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := r.updateWorkspaceSourceBranchesTx(ctx, tx, batch.RepositoryUpdates); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -106,6 +110,31 @@ func (r *Repository) createWorkspaceSourceBatchTx(ctx context.Context, tx *sqlx.
|
||||
return nil
|
||||
}
|
||||
|
||||
// guardWorkspaceSourceParentTx makes cross-task attachment authorization part
|
||||
// of the durable batch transaction. The no-op update takes the target task row
|
||||
// lock on PostgreSQL and SQLite's writer lock, so a reparent either commits
|
||||
// before this predicate (and is rejected) or after the source batch commits.
|
||||
func (r *Repository) guardWorkspaceSourceParentTx(ctx context.Context, tx *sqlx.Tx, batch *models.WorkspaceSourceBatch) error {
|
||||
if batch.ExpectedParentID == "" {
|
||||
return nil
|
||||
}
|
||||
result, err := tx.ExecContext(ctx, r.db.Rebind(`
|
||||
UPDATE tasks SET updated_at = updated_at
|
||||
WHERE id = ? AND parent_id = ? AND workspace_id = ?
|
||||
`), batch.TaskID, batch.ExpectedParentID, batch.ExpectedParentWorkspaceID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("guard workspace source parent relation: %w", err)
|
||||
}
|
||||
changed, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
return fmt.Errorf("guard workspace source parent relation rows affected: %w", err)
|
||||
}
|
||||
if changed == 0 {
|
||||
return fmt.Errorf("%w: %s", repoerrors.ErrTaskParentMismatch, batch.TaskID)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Repository) updateWorkspaceSourceBranchesTx(ctx context.Context, tx *sqlx.Tx, updates []models.WorkspaceSourceRepositoryUpdate) error {
|
||||
for _, update := range updates {
|
||||
if _, err := tx.ExecContext(ctx, r.db.Rebind(`UPDATE task_repositories SET base_branch = ?, updated_at = ? WHERE id = ?`), update.BaseBranch, time.Now().UTC(), update.TaskRepositoryID); err != nil {
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
package sqlite
|
||||
|
||||
// PostgreSQL coverage for the workspace-source parent guard. The guard uses
|
||||
// a no-op UPDATE and RowsAffected to hold the target task row lock until the
|
||||
// source transaction commits, so SQLite-only coverage cannot prove the
|
||||
// cross-connection serialization behavior.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
"github.com/kandev/kandev/internal/task/repository/repoerrors"
|
||||
"github.com/kandev/kandev/internal/testutil"
|
||||
)
|
||||
|
||||
func TestPostgresWorkspaceSourceParentGuard(t *testing.T) {
|
||||
db := openIsolatedPostgresMultiConn(t, testutil.PostgresDSNFromEnv(t), 2)
|
||||
repo, err := NewWithDB(db, db, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("init postgres schema: %v", err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
seedWorkspace(t, repo, "workspace-parent-guard-pg")
|
||||
|
||||
t.Run("matching and stale predicates", func(t *testing.T) {
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: "task-parent-guard-pg", WorkspaceID: "workspace-parent-guard-pg", ParentID: "parent-a", Title: "Child",
|
||||
}); err != nil {
|
||||
t.Fatalf("seed task: %v", err)
|
||||
}
|
||||
|
||||
if err := repo.CreateWorkspaceSourceBatch(ctx, &models.WorkspaceSourceBatch{
|
||||
TaskID: "task-parent-guard-pg", ExpectedParentID: "parent-a", ExpectedParentWorkspaceID: "workspace-parent-guard-pg",
|
||||
Sources: []models.WorkspaceSource{{Folder: &models.TaskWorkspaceFolder{LocalPath: "/canonical/approved", DisplayName: "approved"}}},
|
||||
}); err != nil {
|
||||
t.Fatalf("matching parent write: %v", err)
|
||||
}
|
||||
|
||||
if err := repo.CreateWorkspaceSourceBatch(ctx, &models.WorkspaceSourceBatch{
|
||||
TaskID: "task-parent-guard-pg", ExpectedParentID: "stale-parent", ExpectedParentWorkspaceID: "workspace-parent-guard-pg",
|
||||
Sources: []models.WorkspaceSource{{Folder: &models.TaskWorkspaceFolder{LocalPath: "/canonical/rejected", DisplayName: "rejected"}}},
|
||||
}); !errors.Is(err, repoerrors.ErrTaskParentMismatch) {
|
||||
t.Fatalf("stale parent error = %v, want parent mismatch", err)
|
||||
}
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, "task-parent-guard-pg")
|
||||
if err != nil {
|
||||
t.Fatalf("list folders: %v", err)
|
||||
}
|
||||
if len(folders) != 1 || folders[0].DisplayName != "approved" {
|
||||
t.Fatalf("folders after stale write = %#v, want only approved", folders)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("source commit serializes reparent", func(t *testing.T) {
|
||||
const taskID = "task-parent-guard-source-wins-pg"
|
||||
if err := repo.CreateTask(ctx, &models.Task{ID: taskID, WorkspaceID: "workspace-parent-guard-pg", ParentID: "parent-a", Title: "Source wins"}); err != nil {
|
||||
t.Fatalf("seed task: %v", err)
|
||||
}
|
||||
|
||||
tx, err := db.BeginTxx(ctx, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("begin source transaction: %v", err)
|
||||
}
|
||||
batch := &models.WorkspaceSourceBatch{
|
||||
TaskID: taskID, ExpectedParentID: "parent-a", ExpectedParentWorkspaceID: "workspace-parent-guard-pg",
|
||||
Sources: []models.WorkspaceSource{{Folder: &models.TaskWorkspaceFolder{LocalPath: "/canonical/source-wins", DisplayName: "source-wins"}}},
|
||||
}
|
||||
if err := repo.createWorkspaceSourceBatchTx(ctx, tx, batch); err != nil {
|
||||
_ = tx.Rollback()
|
||||
t.Fatalf("prepare source transaction: %v", err)
|
||||
}
|
||||
|
||||
reparentStarted := make(chan struct{})
|
||||
reparentDone := make(chan error, 1)
|
||||
go func() {
|
||||
close(reparentStarted)
|
||||
_, updateErr := db.ExecContext(ctx, db.Rebind(`UPDATE tasks SET parent_id = ? WHERE id = ?`), "parent-b", taskID)
|
||||
reparentDone <- updateErr
|
||||
}()
|
||||
<-reparentStarted
|
||||
assertBlockedUntilCommit(t, reparentDone, "reparent")
|
||||
if err := tx.Commit(); err != nil {
|
||||
t.Fatalf("commit source transaction: %v", err)
|
||||
}
|
||||
if err := waitForConcurrentResult(reparentDone, "reparent"); err != nil {
|
||||
t.Fatalf("reparent after source commit: %v", err)
|
||||
}
|
||||
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, taskID)
|
||||
if err != nil {
|
||||
t.Fatalf("list source-wins folders: %v", err)
|
||||
}
|
||||
if len(folders) != 1 || folders[0].DisplayName != "source-wins" {
|
||||
t.Fatalf("source-wins folders = %#v, want one committed folder", folders)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("reparent commit rejects stale source", func(t *testing.T) {
|
||||
const taskID = "task-parent-guard-reparent-wins-pg"
|
||||
if err := repo.CreateTask(ctx, &models.Task{ID: taskID, WorkspaceID: "workspace-parent-guard-pg", ParentID: "parent-a", Title: "Reparent wins"}); err != nil {
|
||||
t.Fatalf("seed task: %v", err)
|
||||
}
|
||||
|
||||
tx, err := db.BeginTxx(ctx, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("begin reparent transaction: %v", err)
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, db.Rebind(`UPDATE tasks SET parent_id = ? WHERE id = ?`), "parent-b", taskID); err != nil {
|
||||
_ = tx.Rollback()
|
||||
t.Fatalf("prepare reparent transaction: %v", err)
|
||||
}
|
||||
|
||||
sourceStarted := make(chan struct{})
|
||||
sourceDone := make(chan error, 1)
|
||||
go func() {
|
||||
close(sourceStarted)
|
||||
sourceDone <- repo.CreateWorkspaceSourceBatch(ctx, &models.WorkspaceSourceBatch{
|
||||
TaskID: taskID, ExpectedParentID: "parent-a", ExpectedParentWorkspaceID: "workspace-parent-guard-pg",
|
||||
Sources: []models.WorkspaceSource{{Folder: &models.TaskWorkspaceFolder{LocalPath: "/canonical/reparent-wins", DisplayName: "reparent-wins"}}},
|
||||
})
|
||||
}()
|
||||
<-sourceStarted
|
||||
assertBlockedUntilCommit(t, sourceDone, "source")
|
||||
if err := tx.Commit(); err != nil {
|
||||
t.Fatalf("commit reparent transaction: %v", err)
|
||||
}
|
||||
if err := waitForConcurrentResult(sourceDone, "source"); !errors.Is(err, repoerrors.ErrTaskParentMismatch) {
|
||||
t.Fatalf("source after reparent commit error = %v, want parent mismatch", err)
|
||||
}
|
||||
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, taskID)
|
||||
if err != nil {
|
||||
t.Fatalf("list reparent-wins folders: %v", err)
|
||||
}
|
||||
if len(folders) != 0 {
|
||||
t.Fatalf("reparent-wins folders = %#v, want no write", folders)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func assertBlockedUntilCommit(t *testing.T, result <-chan error, operation string) {
|
||||
t.Helper()
|
||||
timer := time.NewTimer(100 * time.Millisecond)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case err := <-result:
|
||||
t.Fatalf("%s completed before the locking transaction committed: %v", operation, err)
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
|
||||
func waitForConcurrentResult(result <-chan error, operation string) error {
|
||||
timer := time.NewTimer(2 * time.Second)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case err := <-result:
|
||||
return err
|
||||
case <-timer.C:
|
||||
return errors.New(operation + " did not complete after the locking transaction committed")
|
||||
}
|
||||
}
|
||||
@@ -2,10 +2,12 @@ package sqlite
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
"github.com/kandev/kandev/internal/task/repository/repoerrors"
|
||||
)
|
||||
|
||||
func TestTaskWorkspaceFoldersSchemaPersistsCanonicalFolders(t *testing.T) {
|
||||
@@ -121,6 +123,45 @@ func TestCreateWorkspaceSourceBatchAllocatesMixedPositionsAndCompensates(t *test
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreateWorkspaceSourceBatchRejectsStaleParentAuthorizationBeforeWrites(t *testing.T) {
|
||||
repo := newRepoForEntityTests(t)
|
||||
ctx := context.Background()
|
||||
seedWorkspace(t, repo, "workspace-parent-guard")
|
||||
if err := repo.CreateTask(ctx, &models.Task{
|
||||
ID: "task-parent-guard", WorkspaceID: "workspace-parent-guard", ParentID: "current-parent", Title: "Child", Priority: "medium",
|
||||
}); err != nil {
|
||||
t.Fatalf("seed child task: %v", err)
|
||||
}
|
||||
|
||||
batch := &models.WorkspaceSourceBatch{
|
||||
TaskID: "task-parent-guard",
|
||||
ExpectedParentID: "stale-parent",
|
||||
ExpectedParentWorkspaceID: "workspace-parent-guard",
|
||||
Sources: []models.WorkspaceSource{{Folder: &models.TaskWorkspaceFolder{
|
||||
LocalPath: "/canonical/docs", DisplayName: "docs",
|
||||
}}},
|
||||
}
|
||||
validBatch := *batch
|
||||
validBatch.ExpectedParentID = "current-parent"
|
||||
validBatch.Sources = []models.WorkspaceSource{{Folder: &models.TaskWorkspaceFolder{
|
||||
LocalPath: "/canonical/approved", DisplayName: "approved",
|
||||
}}}
|
||||
if err := repo.CreateWorkspaceSourceBatch(ctx, &validBatch); err != nil {
|
||||
t.Fatalf("CreateWorkspaceSourceBatch matching parent: %v", err)
|
||||
}
|
||||
err := repo.CreateWorkspaceSourceBatch(ctx, batch)
|
||||
if !errors.Is(err, repoerrors.ErrTaskParentMismatch) {
|
||||
t.Fatalf("CreateWorkspaceSourceBatch error = %v, want stale parent authorization", err)
|
||||
}
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, batch.TaskID)
|
||||
if err != nil {
|
||||
t.Fatalf("list folders: %v", err)
|
||||
}
|
||||
if len(folders) != 1 || folders[0].DisplayName != "approved" {
|
||||
t.Fatalf("folders = %#v, want only matching-parent write", folders)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskWorkspaceFolderRejectsDuplicatePathOrDisplayNameAndCascades(t *testing.T) {
|
||||
repo := newRepoForEntityTests(t)
|
||||
ctx := context.Background()
|
||||
|
||||
@@ -33,8 +33,10 @@ type WorkspaceSourceInput struct {
|
||||
}
|
||||
|
||||
type AttachWorkspaceSourcesRequest struct {
|
||||
TaskID string
|
||||
Sources []WorkspaceSourceInput
|
||||
TaskID string
|
||||
Sources []WorkspaceSourceInput
|
||||
ExpectedParentID string
|
||||
ExpectedParentWorkspaceID string
|
||||
}
|
||||
|
||||
// AttachWorkspaceSourcesResult is the authoritative post-materialization
|
||||
@@ -45,6 +47,9 @@ type AttachWorkspaceSourcesResult struct {
|
||||
WorkspacePath string
|
||||
WorktreePath string
|
||||
SessionIDs []string
|
||||
// Changed reports whether this request made a durable mutation. Exact
|
||||
// normalized retries return the same projection without runtime side effects.
|
||||
Changed bool
|
||||
}
|
||||
|
||||
// WorkspaceSourceMaterializationResult is returned by the runtime boundary
|
||||
@@ -94,10 +99,10 @@ func (s *Service) AttachWorkspaceSources(ctx context.Context, req AttachWorkspac
|
||||
if task == nil {
|
||||
return nil, fmt.Errorf("%w: %s", taskrepository.ErrTaskNotFound, req.TaskID)
|
||||
}
|
||||
if err := s.workspaceSourcesIdle(ctx, task.ID); err != nil {
|
||||
return nil, err
|
||||
if req.ExpectedParentID != "" && (task.ParentID != req.ExpectedParentID || task.WorkspaceID != req.ExpectedParentWorkspaceID) {
|
||||
return nil, fmt.Errorf("%w: %s", taskrepository.ErrTaskParentMismatch, task.ID)
|
||||
}
|
||||
if err := s.rejectUnsupportedFolderSources(ctx, task.ID, req.Sources); err != nil {
|
||||
if err := s.workspaceSourcesIdle(ctx, task.ID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
existing, err := s.taskRepos.ListTaskRepositories(ctx, task.ID)
|
||||
@@ -115,13 +120,101 @@ func (s *Service) AttachWorkspaceSources(ctx context.Context, req AttachWorkspac
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
batch, cleanupCreated, err := s.prepareWorkspaceSourceBatch(ctx, task, existing, folders, req.Sources)
|
||||
inputs, err := filterExactWorkspaceSourceDuplicates(existing, folders, req.Sources)
|
||||
if err != nil {
|
||||
// Preserve the capability-first error for unsupported executors when a
|
||||
// new folder path cannot be normalized. Valid paths still reach the
|
||||
// duplicate filter first, so exact retries remain no-ops after an
|
||||
// executor change.
|
||||
if errors.Is(err, ErrInvalidWorkspaceSource) {
|
||||
if capabilityErr := s.rejectUnsupportedFolderSources(ctx, task.ID, req.Sources); capabilityErr != nil {
|
||||
return nil, capabilityErr
|
||||
}
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if err := s.rejectUnsupportedFolderSources(ctx, task.ID, inputs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(inputs) == 0 {
|
||||
if err := guardWorkspaceSourceParent(ctx, store, task, req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.hydrateWorkspaceSourceResult(ctx, task, store)
|
||||
}
|
||||
batch, cleanupCreated, err := s.prepareWorkspaceSourceBatch(ctx, task, existing, folders, inputs)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
batch.ExpectedParentID = req.ExpectedParentID
|
||||
batch.ExpectedParentWorkspaceID = req.ExpectedParentWorkspaceID
|
||||
if len(batch.Sources) == 0 && len(batch.RepositoryUpdates) == 0 {
|
||||
cleanupCreated(context.WithoutCancel(ctx))
|
||||
if err := guardWorkspaceSourceParent(ctx, store, task, req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return s.hydrateWorkspaceSourceResult(ctx, task, store)
|
||||
}
|
||||
return s.commitWorkspaceSourceBatch(ctx, task, batch, cleanupCreated, s.materializeWorkspaceSources)
|
||||
}
|
||||
|
||||
func guardWorkspaceSourceParent(ctx context.Context, store taskrepository.TaskWorkspaceFolderRepository, task *models.Task, req AttachWorkspaceSourcesRequest) error {
|
||||
if req.ExpectedParentID == "" {
|
||||
return nil
|
||||
}
|
||||
return store.CreateWorkspaceSourceBatch(ctx, &models.WorkspaceSourceBatch{
|
||||
TaskID: task.ID,
|
||||
ExpectedParentID: req.ExpectedParentID,
|
||||
ExpectedParentWorkspaceID: req.ExpectedParentWorkspaceID,
|
||||
})
|
||||
}
|
||||
|
||||
func filterExactWorkspaceSourceDuplicates(existing []*models.TaskRepository, folders []*models.TaskWorkspaceFolder, inputs []WorkspaceSourceInput) ([]WorkspaceSourceInput, error) {
|
||||
paths, names := map[string]string{}, map[string]string{}
|
||||
for _, folder := range folders {
|
||||
paths[folder.LocalPath] = folder.DisplayName
|
||||
names[folder.DisplayName] = folder.LocalPath
|
||||
}
|
||||
filtered := make([]WorkspaceSourceInput, 0, len(inputs))
|
||||
for _, input := range inputs {
|
||||
if input.Kind == WorkspaceSourceFolder {
|
||||
folder, duplicate, err := filterWorkspaceFolderDuplicate(input, paths, names)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !duplicate {
|
||||
filtered = append(filtered, folder)
|
||||
}
|
||||
continue
|
||||
}
|
||||
filtered = append(filtered, input)
|
||||
}
|
||||
return filtered, nil
|
||||
}
|
||||
|
||||
func filterWorkspaceFolderDuplicate(input WorkspaceSourceInput, paths, names map[string]string) (WorkspaceSourceInput, bool, error) {
|
||||
path, err := canonicalFolder(input.LocalPath)
|
||||
if err != nil {
|
||||
return input, false, fmt.Errorf("%w: %v", ErrInvalidWorkspaceSource, err)
|
||||
}
|
||||
name := input.DisplayName
|
||||
if name == "" {
|
||||
name = filepath.Base(path)
|
||||
}
|
||||
if existingName, exists := paths[path]; exists {
|
||||
if existingName == name {
|
||||
return input, true, nil
|
||||
}
|
||||
return input, false, fmt.Errorf("%w: workspace folder path is already attached", ErrWorkspaceSourceConflict)
|
||||
}
|
||||
if existingPath, exists := names[name]; exists && existingPath != path {
|
||||
return input, false, fmt.Errorf("%w: workspace folder name is already attached", ErrWorkspaceSourceConflict)
|
||||
}
|
||||
paths[path], names[name] = name, path
|
||||
input.LocalPath, input.DisplayName = path, name
|
||||
return input, false, nil
|
||||
}
|
||||
|
||||
func (s *Service) prepareWorkspaceSourceBatch(ctx context.Context, task *models.Task, existing []*models.TaskRepository, folders []*models.TaskWorkspaceFolder, inputs []WorkspaceSourceInput) (*models.WorkspaceSourceBatch, func(context.Context), error) {
|
||||
batch := &models.WorkspaceSourceBatch{TaskID: task.ID}
|
||||
created := make([]string, 0)
|
||||
@@ -151,10 +244,16 @@ func (s *Service) prepareWorkspaceSourceBatch(ctx context.Context, task *models.
|
||||
}
|
||||
batch.Sources = append(batch.Sources, models.WorkspaceSource{Folder: folder})
|
||||
case WorkspaceSourceRepository:
|
||||
taskRepository, createdByUs, err := s.prepareRepositoryWorkspaceSource(ctx, task, input, prospective)
|
||||
taskRepository, createdByUs, duplicate, err := s.prepareRepositoryWorkspaceSource(ctx, task, input, prospective)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if duplicate {
|
||||
if createdByUs != "" {
|
||||
s.cleanupCreatedWorkspaceRepositories(context.WithoutCancel(ctx), []string{createdByUs})
|
||||
}
|
||||
continue
|
||||
}
|
||||
if createdByUs != "" {
|
||||
created = append(created, createdByUs)
|
||||
}
|
||||
@@ -230,39 +329,57 @@ func prepareFolderWorkspaceSource(input WorkspaceSourceInput, seenPaths, seenNam
|
||||
return &models.TaskWorkspaceFolder{LocalPath: path, DisplayName: name}, nil
|
||||
}
|
||||
|
||||
func (s *Service) prepareRepositoryWorkspaceSource(ctx context.Context, task *models.Task, input WorkspaceSourceInput, existing []*models.TaskRepository) (*models.TaskRepository, string, error) {
|
||||
func (s *Service) prepareRepositoryWorkspaceSource(ctx context.Context, task *models.Task, input WorkspaceSourceInput, existing []*models.TaskRepository) (*models.TaskRepository, string, bool, error) {
|
||||
if err := s.validateRepositoryWorkspaceSourceInput(ctx, task, input); err != nil {
|
||||
return nil, "", err
|
||||
return nil, "", false, err
|
||||
}
|
||||
id, base, createdID, err := s.resolveRepositoryWorkspaceSource(ctx, task, input)
|
||||
if err != nil {
|
||||
return nil, createdID, err
|
||||
return nil, createdID, false, err
|
||||
}
|
||||
repo, base, err := s.resolveWorkspaceSourceBaseBranch(ctx, task, id, base)
|
||||
if err != nil {
|
||||
return nil, createdID, err
|
||||
return nil, createdID, false, err
|
||||
}
|
||||
if err := validateWorkspaceSourceBranches(base, input.CheckoutBranch); err != nil {
|
||||
return nil, createdID, err
|
||||
return nil, createdID, false, err
|
||||
}
|
||||
if input.LocalPath != "" {
|
||||
if err := s.requireCloneableLocalRepository(ctx, task.ID, repo); err != nil {
|
||||
return nil, createdID, err
|
||||
return nil, createdID, false, err
|
||||
}
|
||||
}
|
||||
if hasExactWorkspaceSourceRepository(existing, id, base, input.CheckoutBranch) {
|
||||
return nil, createdID, true, nil
|
||||
}
|
||||
checkout := input.CheckoutBranch
|
||||
if _, duplicate := scanForBranchAddDuplicate(existing, id, base, checkout, repo); duplicate != nil {
|
||||
return nil, createdID, fmt.Errorf("%w: %v", ErrWorkspaceSourceConflict, duplicate)
|
||||
return nil, createdID, false, fmt.Errorf("%w: %v", ErrWorkspaceSourceConflict, duplicate)
|
||||
}
|
||||
return &models.TaskRepository{RepositoryID: id, BaseBranch: base, CheckoutBranch: checkout, Metadata: map[string]interface{}{}}, createdID, nil
|
||||
return &models.TaskRepository{RepositoryID: id, BaseBranch: base, CheckoutBranch: checkout, Metadata: map[string]interface{}{}}, createdID, false, nil
|
||||
}
|
||||
|
||||
func hasExactWorkspaceSourceRepository(existing []*models.TaskRepository, repositoryID, baseBranch, checkoutBranch string) bool {
|
||||
for _, repository := range existing {
|
||||
if repository != nil && repository.RepositoryID == repositoryID && repository.BaseBranch == baseBranch && repository.CheckoutBranch == checkoutBranch {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (s *Service) validateRepositoryWorkspaceSourceInput(ctx context.Context, task *models.Task, input WorkspaceSourceInput) error {
|
||||
if repositoryLocatorCount(input) != 1 {
|
||||
return fmt.Errorf("%w: repository source requires exactly one locator", ErrInvalidWorkspaceSource)
|
||||
}
|
||||
if err := validateWorkspaceSourceBranches(input.BaseBranch, input.CheckoutBranch); err != nil {
|
||||
return err
|
||||
// base_branch may be omitted when the resolved repository has a default
|
||||
// branch. Validate a supplied value here and the effective value after
|
||||
// resolution in prepareRepositoryWorkspaceSource.
|
||||
if input.BaseBranch != "" && !securityutil.IsValidBranchName(input.BaseBranch) {
|
||||
return fmt.Errorf("%w: base_branch %q is not a safe git branch", ErrInvalidWorkspaceSource, input.BaseBranch)
|
||||
}
|
||||
if input.CheckoutBranch != "" && !securityutil.IsValidBranchName(input.CheckoutBranch) {
|
||||
return fmt.Errorf("%w: checkout_branch %q is not a safe git branch", ErrInvalidWorkspaceSource, input.CheckoutBranch)
|
||||
}
|
||||
if input.RepositoryID == "" {
|
||||
return nil
|
||||
@@ -299,6 +416,9 @@ func (s *Service) resolveWorkspaceSourceBaseBranch(ctx context.Context, task *mo
|
||||
return nil, "", fmt.Errorf("%w: resolve local repository branch: %v", ErrInvalidWorkspaceSource, err)
|
||||
}
|
||||
}
|
||||
if base == "" {
|
||||
base = repo.DefaultBranch
|
||||
}
|
||||
if base == "" {
|
||||
return nil, "", fmt.Errorf("%w: repository source requires base_branch", ErrInvalidWorkspaceSource)
|
||||
}
|
||||
@@ -476,6 +596,7 @@ func (s *Service) commitWorkspaceSourceBatch(ctx context.Context, task *models.T
|
||||
s.PublishWorkspaceSourcesAdopted(context.WithoutCancel(ctx), task.ID, result.WorkspacePath, result.SessionIDs)
|
||||
}
|
||||
succeeded = true
|
||||
result.Changed = true
|
||||
return result, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@ package service
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -11,6 +12,8 @@ import (
|
||||
"github.com/kandev/kandev/internal/events"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
taskrepository "github.com/kandev/kandev/internal/task/repository"
|
||||
sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestAttachWorkspaceSourcesDerivesLegacyPrimaryBranchForWorktreeProjection(t *testing.T) {
|
||||
@@ -855,7 +858,7 @@ func TestAttachWorkspaceSources_RejectsSameRepositoryDifferentBaseOnLocalRuntime
|
||||
}
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesRejectsDuplicateRepositoryWithinBatch(t *testing.T) {
|
||||
func TestAttachWorkspaceSourcesCollapsesDuplicateRepositoryWithinBatch(t *testing.T) {
|
||||
svc, _, repo := createTestService(t)
|
||||
svc.workspaceFolders = repo
|
||||
ctx := context.Background()
|
||||
@@ -874,9 +877,261 @@ func TestAttachWorkspaceSourcesRejectsDuplicateRepositoryWithinBatch(t *testing.
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{{Kind: WorkspaceSourceRepository, RepositoryID: "repo-dup", BaseBranch: "release", CheckoutBranch: "feature/x"}, {Kind: WorkspaceSourceRepository, RepositoryID: "repo-dup", BaseBranch: "release", CheckoutBranch: "feature/x"}}})
|
||||
if err == nil {
|
||||
t.Fatal("AttachWorkspaceSources succeeded, want duplicate error")
|
||||
if err != nil {
|
||||
t.Fatalf("AttachWorkspaceSources: %v", err)
|
||||
}
|
||||
rows, err := repo.ListTaskRepositories(ctx, task.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 2 {
|
||||
t.Fatalf("task repositories = %d, want primary plus one attachment", len(rows))
|
||||
}
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesExactRetriesSkipRuntimeSideEffects(t *testing.T) {
|
||||
svc, eventBus, repo := createTestService(t)
|
||||
svc.workspaceFolders = repo
|
||||
ctx := context.Background()
|
||||
if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-idempotent-retry", Name: "Retry"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-idempotent-retry", WorkspaceID: "ws-idempotent-retry", Name: "WF"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := repo.CreateRepository(ctx, &models.Repository{ID: "repo-idempotent-retry", WorkspaceID: "ws-idempotent-retry", Name: "app", DefaultBranch: "main"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
taskResult, err := svc.CreateTask(ctx, &CreateTaskRequest{
|
||||
WorkspaceID: "ws-idempotent-retry", WorkflowID: "wf-idempotent-retry", Title: "Task",
|
||||
Repositories: []TaskRepositoryInput{{RepositoryID: "repo-idempotent-retry", BaseBranch: "main"}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task := taskResult.Task
|
||||
source := WorkspaceSourceInput{Kind: WorkspaceSourceRepository, RepositoryID: "repo-idempotent-retry", BaseBranch: "release", CheckoutBranch: "feature/retry"}
|
||||
if _, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}}); err != nil {
|
||||
t.Fatalf("initial attachment: %v", err)
|
||||
}
|
||||
eventBus.ClearEvents()
|
||||
materializer := &recordingWorkspaceSourceMaterializer{}
|
||||
refresher := &recordingWorkspaceSourceProviderRefresher{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
|
||||
result, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}})
|
||||
if err != nil {
|
||||
t.Fatalf("exact retry: %v", err)
|
||||
}
|
||||
if result.Changed || materializer.called || refresher.calls != 0 || len(eventBus.GetPublishedEvents()) != 0 {
|
||||
t.Fatalf("exact retry changed=%t materialized=%t refreshes=%d events=%d, want no side effects", result.Changed, materializer.called, refresher.calls, len(eventBus.GetPublishedEvents()))
|
||||
}
|
||||
rows, err := repo.ListTaskRepositories(ctx, task.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(rows) != 2 {
|
||||
t.Fatalf("task repositories after exact retry = %d, want 2", len(rows))
|
||||
}
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesExactFolderRetryCanonicalizesPathWithoutSideEffects(t *testing.T) {
|
||||
svc, eventBus, repo, task := newWorkspaceSourceRetryFixture(t)
|
||||
ctx := context.Background()
|
||||
folder := t.TempDir()
|
||||
source := WorkspaceSourceInput{Kind: WorkspaceSourceFolder, LocalPath: folder, DisplayName: "docs"}
|
||||
_, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}})
|
||||
require.NoError(t, err)
|
||||
|
||||
eventBus.ClearEvents()
|
||||
materializer := &recordingWorkspaceSourceMaterializer{}
|
||||
refresher := &recordingWorkspaceSourceProviderRefresher{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
retry, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{{
|
||||
Kind: WorkspaceSourceFolder, LocalPath: workspaceSourceTestSymlink(t, folder), DisplayName: "docs",
|
||||
}}})
|
||||
require.NoError(t, err)
|
||||
require.False(t, retry.Changed)
|
||||
require.False(t, materializer.called)
|
||||
require.Zero(t, refresher.calls)
|
||||
require.Empty(t, eventBus.GetPublishedEvents())
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, task.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, folders, 1)
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesExactFolderRetryIgnoresExecutorChange(t *testing.T) {
|
||||
svc, _, repo, task := newWorkspaceSourceRetryFixture(t)
|
||||
ctx := context.Background()
|
||||
folder := t.TempDir()
|
||||
source := WorkspaceSourceInput{Kind: WorkspaceSourceFolder, LocalPath: folder, DisplayName: "docs"}
|
||||
_, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}})
|
||||
require.NoError(t, err)
|
||||
|
||||
// A task can switch executor profiles between launches. An exact retry must
|
||||
// remain an idempotent no-op even when the new executor cannot accept folders.
|
||||
require.NoError(t, repo.CreateTaskEnvironment(ctx, &models.TaskEnvironment{
|
||||
ID: "env-retry-remote", TaskID: task.ID, ExecutorType: "remote_docker",
|
||||
}))
|
||||
retry, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}})
|
||||
require.NoError(t, err)
|
||||
require.False(t, retry.Changed)
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, task.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, folders, 1)
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesMixedDuplicateAndNewFoldersCommitsOnlyNewSource(t *testing.T) {
|
||||
svc, eventBus, repo, task := newWorkspaceSourceRetryFixture(t)
|
||||
ctx := context.Background()
|
||||
existing := t.TempDir()
|
||||
newFolder := t.TempDir()
|
||||
_, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{{
|
||||
Kind: WorkspaceSourceFolder, LocalPath: existing, DisplayName: "existing",
|
||||
}}})
|
||||
require.NoError(t, err)
|
||||
|
||||
eventBus.ClearEvents()
|
||||
materializer := &recordingWorkspaceSourceMaterializer{}
|
||||
refresher := &recordingWorkspaceSourceProviderRefresher{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
result, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{
|
||||
{Kind: WorkspaceSourceFolder, LocalPath: workspaceSourceTestSymlink(t, existing), DisplayName: "existing"},
|
||||
{Kind: WorkspaceSourceFolder, LocalPath: newFolder, DisplayName: "new"},
|
||||
}})
|
||||
require.NoError(t, err)
|
||||
require.True(t, result.Changed)
|
||||
require.True(t, materializer.called)
|
||||
require.Equal(t, 1, refresher.calls)
|
||||
require.NotEmpty(t, eventBus.GetPublishedEvents())
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, task.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, folders, 2)
|
||||
require.Equal(t, "existing", folders[0].DisplayName)
|
||||
require.Equal(t, "new", folders[1].DisplayName)
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesRejectsContradictoryFolderDuplicates(t *testing.T) {
|
||||
svc, eventBus, repo, task := newWorkspaceSourceRetryFixture(t)
|
||||
ctx := context.Background()
|
||||
folder := t.TempDir()
|
||||
_, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{{
|
||||
Kind: WorkspaceSourceFolder, LocalPath: folder, DisplayName: "docs",
|
||||
}}})
|
||||
require.NoError(t, err)
|
||||
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
source WorkspaceSourceInput
|
||||
}{
|
||||
{name: "same path different name", source: WorkspaceSourceInput{Kind: WorkspaceSourceFolder, LocalPath: workspaceSourceTestSymlink(t, folder), DisplayName: "renamed"}},
|
||||
{name: "same name different path", source: WorkspaceSourceInput{Kind: WorkspaceSourceFolder, LocalPath: t.TempDir(), DisplayName: "docs"}},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
eventBus.ClearEvents()
|
||||
result, attachErr := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{testCase.source}})
|
||||
require.ErrorIs(t, attachErr, ErrWorkspaceSourceConflict)
|
||||
require.Nil(t, result)
|
||||
require.Empty(t, eventBus.GetPublishedEvents())
|
||||
folders, listErr := repo.ListTaskWorkspaceFolders(ctx, task.ID)
|
||||
require.NoError(t, listErr)
|
||||
require.Len(t, folders, 1)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func workspaceSourceTestSymlink(t *testing.T, target string) string {
|
||||
t.Helper()
|
||||
path := filepath.Join(t.TempDir(), "folder-link")
|
||||
if err := os.Symlink(target, path); err != nil {
|
||||
t.Skipf("create workspace-source test symlink: %v", err)
|
||||
}
|
||||
return path
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesExactRepositoryRetryNormalizesDefaultBaseBranch(t *testing.T) {
|
||||
svc, eventBus, repo, task := newWorkspaceSourceRetryFixture(t)
|
||||
ctx := context.Background()
|
||||
require.NoError(t, repo.CreateRepository(ctx, &models.Repository{ID: "repo-retry-added", WorkspaceID: "ws-retry", Name: "added", DefaultBranch: "main"}))
|
||||
source := WorkspaceSourceInput{Kind: WorkspaceSourceRepository, RepositoryID: "repo-retry-added", CheckoutBranch: "feature/retry"}
|
||||
_, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}})
|
||||
require.NoError(t, err)
|
||||
|
||||
eventBus.ClearEvents()
|
||||
materializer := &recordingWorkspaceSourceMaterializer{}
|
||||
refresher := &recordingWorkspaceSourceProviderRefresher{}
|
||||
svc.SetWorkspaceSourceMaterializer(materializer)
|
||||
svc.SetWorkspaceSourceProviderRefresher(refresher)
|
||||
retry, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{{
|
||||
Kind: WorkspaceSourceRepository, RepositoryID: "repo-retry-added", BaseBranch: "main", CheckoutBranch: "feature/retry",
|
||||
}}})
|
||||
require.NoError(t, err)
|
||||
require.False(t, retry.Changed)
|
||||
require.False(t, materializer.called)
|
||||
require.Zero(t, refresher.calls)
|
||||
require.Empty(t, eventBus.GetPublishedEvents())
|
||||
repositories, err := repo.ListTaskRepositories(ctx, task.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, repositories, 2)
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesConcurrentExactFolderRetriesRemainNoOps(t *testing.T) {
|
||||
svc, eventBus, repo, task := newWorkspaceSourceRetryFixture(t)
|
||||
ctx := context.Background()
|
||||
folder := t.TempDir()
|
||||
source := WorkspaceSourceInput{Kind: WorkspaceSourceFolder, LocalPath: folder, DisplayName: "docs"}
|
||||
_, err := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}})
|
||||
require.NoError(t, err)
|
||||
eventBus.ClearEvents()
|
||||
|
||||
results := make(chan *AttachWorkspaceSourcesResult, 2)
|
||||
errors := make(chan error, 2)
|
||||
start := make(chan struct{})
|
||||
var wait sync.WaitGroup
|
||||
for range 2 {
|
||||
wait.Add(1)
|
||||
go func() {
|
||||
defer wait.Done()
|
||||
<-start
|
||||
result, attachErr := svc.AttachWorkspaceSources(ctx, AttachWorkspaceSourcesRequest{TaskID: task.ID, Sources: []WorkspaceSourceInput{source}})
|
||||
results <- result
|
||||
errors <- attachErr
|
||||
}()
|
||||
}
|
||||
close(start)
|
||||
wait.Wait()
|
||||
close(results)
|
||||
close(errors)
|
||||
for attachErr := range errors {
|
||||
require.NoError(t, attachErr)
|
||||
}
|
||||
for result := range results {
|
||||
require.NotNil(t, result)
|
||||
require.False(t, result.Changed)
|
||||
}
|
||||
require.Empty(t, eventBus.GetPublishedEvents())
|
||||
folders, err := repo.ListTaskWorkspaceFolders(ctx, task.ID)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, folders, 1)
|
||||
}
|
||||
|
||||
func newWorkspaceSourceRetryFixture(t *testing.T) (*Service, *MockEventBus, *sqliterepo.Repository, *models.Task) {
|
||||
t.Helper()
|
||||
svc, eventBus, repo := createTestService(t)
|
||||
svc.workspaceFolders = repo
|
||||
ctx := context.Background()
|
||||
require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-retry", Name: "Retry"}))
|
||||
require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-retry", WorkspaceID: "ws-retry", Name: "Workflow"}))
|
||||
require.NoError(t, repo.CreateRepository(ctx, &models.Repository{ID: "repo-retry-primary", WorkspaceID: "ws-retry", Name: "primary", DefaultBranch: "main"}))
|
||||
created, err := svc.CreateTask(ctx, &CreateTaskRequest{
|
||||
WorkspaceID: "ws-retry", WorkflowID: "wf-retry", WorkflowStepID: "step", Title: "Retry task",
|
||||
Repositories: []TaskRepositoryInput{{RepositoryID: "repo-retry-primary", BaseBranch: "main"}},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
return svc, eventBus, repo, created.Task
|
||||
}
|
||||
|
||||
func TestAttachWorkspaceSourcesRejectsSanitizedBranchCollisionWithinBatch(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
# ADR-2026-08-19-parent-authorized-child-workspace-sources: Parent-authorized child workspace sources
|
||||
|
||||
**Status:** accepted
|
||||
**Date:** 2026-08-19
|
||||
**Area:** backend, protocol, security
|
||||
|
||||
## Context
|
||||
|
||||
An idle child can be blocked by a missing repository or SDK and therefore cannot repair its own
|
||||
workspace. The task-mode MCP transport used to reject every explicit target before the backend
|
||||
could apply the task-tree authorization policy, while broad same-workspace access would let
|
||||
unrelated tasks grant one another source access.
|
||||
|
||||
## Decision
|
||||
|
||||
Task-bound MCP may attach workspace sources to the calling task or to its direct child, provided both tasks belong to the same non-empty workspace. The MCP server injects its bound caller task and session IDs; these provenance fields are not callable tool arguments. The backend verifies that the session belongs to the caller task, then carries the direct-parent and workspace predicates into the source-batch transaction. A target reparented before that transaction commits is rejected before any source, repository, folder, session, or event mutation becomes durable.
|
||||
|
||||
Exact normalized retries are no-ops after the idle check. They return the authoritative projection without rematerializing sources, refreshing providers, or publishing update events.
|
||||
|
||||
## Consequences
|
||||
|
||||
This permits a coordinator parent to repair an idle child's missing repository or SDK without granting sibling, ancestor, descendant, cross-workspace, or generic same-owner mutation. `add_branch_to_task_kandev` remains current-task-only because its active-turn live-rescan contract is intentionally different.
|
||||
|
||||
## Alternatives Considered
|
||||
|
||||
Keeping the tool self-only leaves blocked children unable to recover. Authorizing in agentctl would duplicate task-tree policy and bypass the backend's authoritative service boundary. Broad same-workspace authority would allow unrelated tasks to mutate one another.
|
||||
@@ -95,6 +95,7 @@ Read individual ADRs for full context. Create new ones via `/record decision` or
|
||||
| 2026-08-15-portable-agent-configuration-bundles | [Use Explicit Portable Agent Configuration Bundles](2026-08-15-portable-agent-configuration-bundles.md) | accepted | backend, frontend, protocol, security | 2026-08-15 |
|
||||
| 2026-08-15-executor-authoritative-model-selection | [Let the Executor Own Model Selection](2026-08-15-executor-authoritative-model-selection.md) | accepted | backend, frontend, protocol, persistence | 2026-08-15 |
|
||||
| 2026-08-18-context-reset-preserves-runtime-configuration | [Preserve ACP Runtime Configuration Across Context Reset](2026-08-18-context-reset-preserves-runtime-configuration.md) | accepted | backend, protocol, workflow | 2026-08-18 |
|
||||
| 2026-08-19-parent-authorized-child-workspace-sources | [Parent-authorized child workspace sources](2026-08-19-parent-authorized-child-workspace-sources.md) | accepted | backend, protocol, security | 2026-08-19 |
|
||||
| 2026-07-28-coarse-running-busy-signal | [Restore Coarse Running Prompt Admission](2026-07-28-coarse-running-busy-signal.md) | accepted | backend, frontend, protocol | 2026-07-28 |
|
||||
| 2026-07-29-agent-stall-user-controlled-recovery | [Keep Agent Stall Recovery User Controlled](2026-07-29-agent-stall-user-controlled-recovery.md) | accepted | backend, frontend, protocol | 2026-07-29 |
|
||||
| 2026-07-29-quarantine-retention-override | [Make Quarantine Retention Overridable but Visible](2026-07-29-quarantine-retention-override.md) | accepted | backend, frontend | 2026-07-29 |
|
||||
|
||||
@@ -523,11 +523,11 @@ Use `stop_task_kandev` only when the direct child should halt without a replacem
|
||||
|
||||
After an accepted stop, Kandev attempts to move an unarchived, non-Office task from `IN_PROGRESS` or `SCHEDULING` to `REVIEW`; other task states are preserved. Worktrees, task environments, commits, task records, descendants, and queued messages remain available, and the task can be started again later.
|
||||
|
||||
`add_workspace_sources_kandev` adds one or more sources to an idle task and defaults `task_id` to the current task. Its `sources` input accepts the same atomic mixed batch as the Files panel: `repository` sources use exactly one saved repository ID, local Git path, or remote repository locator plus branch fields; `folder` sources use a local path and optional display name. Repository sources work on Worktree, Local/Local PC, Local Docker, SSH, and Sprites; folders work only on Worktree and Local/Local PC. The task must be repository-backed and have no active turn or tool call. Invalid, duplicate, unsupported, or failed sources roll back the entire batch.
|
||||
`add_workspace_sources_kandev` adds one or more sources to an idle task and defaults `task_id` to the current task. A task may also target its same-workspace direct child; Kandev verifies the calling task and session on the backend, so agents cannot provide or override that provenance. Its `sources` input accepts the same atomic mixed batch as the Files panel: `repository` sources use exactly one saved repository ID, local Git path, or remote repository locator plus branch fields; `folder` sources use a local path and optional display name. Repository sources work on Worktree, Local/Local PC, Local Docker, SSH, and Sprites; folders work only on Worktree and Local/Local PC. The target must be repository-backed and have no active turn or tool call. Exact normalized retries are idempotent; contradictory duplicates, unsupported, or failed sources roll back the batch.
|
||||
|
||||
`add_branch_to_task_kandev` is the Worktree-only compatibility path for adding one repository/branch during an active agent turn. It creates the worktree as a sibling under the task directory, promotes the persisted Files root to that parent, and rescans it without restarting the agent, terminals, or workspace processes. The response returns `worktree_path` (the exact new repository location), `task_workspace_path` (the Files root), and `agent_cwd_changed: false`; deferred pre-launch materialization omits both paths. The original repository stays a separate Git worktree, so the sibling is not reported as an embedded repository or untracked files by its Git status. Use `add_workspace_sources_kandev` for mixed batch attachments to an idle task. `update_repository_base_branch_kandev` changes the base used for Kandev's diff, not a pull request's target branch.
|
||||
|
||||
The HTTP equivalent is `POST /api/v1/tasks/:id/workspace-sources`, with `{ "sources": [...] }`. It returns `400` for invalid input, `404` for a missing task/source outside the workspace, `409` for duplicates or an active task, and `422` when materialization or executor capability fails. Successful adoption publishes `task.updated` and `session.workspace_sources.updated`; clients should refresh their Files and repository state from those updates.
|
||||
The HTTP equivalent is `POST /api/v1/tasks/:id/workspace-sources`, with `{ "sources": [...] }`. An exact normalized retry succeeds as a no-op. It returns `400` for invalid input, `404` for a missing task/source outside the workspace, `409` for contradictory duplicates or an active task, and `422` when materialization or executor capability fails. Successful adoption publishes `task.updated` and `session.workspace_sources.updated`; clients should refresh their Files and repository state from those updates.
|
||||
|
||||
`step_complete_kandev` is registered and discoverable in every task-mode session. Kandev includes its completion instruction, and acts on its signal, only on Kanban steps whose auto-advance action explicitly requires that signal. A user message arriving before transition can cancel that automatic move.
|
||||
|
||||
|
||||
@@ -243,7 +243,7 @@ processes after attachment. Local Docker, SSH, and Sprites instead clone the new
|
||||
the current remote workspace and rescan it without changing the agent CWD or restarting the agent
|
||||
and workspace processes.
|
||||
|
||||
Task agents can call `add_workspace_sources_kandev` with the same mixed batch; `task_id` defaults to the current task, and the operation remains idle-only. `add_branch_to_task_kandev` is the Worktree-only legacy one-repository/branch path and may run during an active turn: it creates a sibling worktree under the task directory, promotes the Files root to that parent, and rescans without restarting the agent, terminals, or workspace processes. Its `worktree_path` is the exact new location, `task_workspace_path` is the Files root, and `agent_cwd_changed` is always `false`; the agent's current directory stays unchanged.
|
||||
Task agents can call `add_workspace_sources_kandev` with the same mixed batch; `task_id` defaults to the current task, and the operation remains idle-only. A direct parent in the same workspace can use it to recover an idle child that is missing a repository or SDK; siblings and other task relationships cannot target that child. Exact retries are safe no-ops. `add_branch_to_task_kandev` remains the current-task-only Worktree legacy one-repository/branch path and may run during an active turn: it creates a sibling worktree under the task directory, promotes the Files root to that parent, and rescans without restarting the agent, terminals, or workspace processes. Its `worktree_path` is the exact new location, `task_workspace_path` is the Files root, and `agent_cwd_changed` is always `false`; the agent's current directory stays unchanged.
|
||||
|
||||
Use `update_repository_base_branch_kandev` with a task-repository ID to change the comparison base. The database update is authoritative. Resetting cached session bases, refreshing Changes, base commit, ahead/behind counts, and cumulative diff in a live tracker are best-effort side effects; a failure is logged without rolling back the new base, and the persisted value is rebuilt on the next session launch. The tool does not rewrite commits, switch the checkout, or change an existing pull request's target branch.
|
||||
|
||||
|
||||
@@ -106,6 +106,8 @@ launching an agent.
|
||||
|
||||
Adding a repository or folder gives the task access to that source. A local folder is a live host-path grant, not an upload: Kandev does not copy, move, delete, or add marker files to it. Folder sources are therefore limited to Local/Local PC and Worktree tasks and are never sent to Docker, SSH, Sprites, or Remote Docker.
|
||||
|
||||
For task MCP source attachment, a task can grant a source to itself or to its direct child in the same workspace only. The backend verifies the session-bound caller identity; sibling, ancestor, non-direct descendant, unrelated, and cross-workspace tasks cannot use this path to grant access. Treat a parent-to-child attachment as a deliberate access grant, and keep the target idle while it is made.
|
||||
|
||||
Remote repository locators and clone credentials can reveal authority. Kandev does not persist credential-bearing URLs or include credentials in source metadata or logs. Use provider credentials or a safe cloneable locator, and never paste tokens into a repository URL, task prompt, or source display name.
|
||||
|
||||
## Protect stored secrets
|
||||
|
||||
@@ -33,7 +33,7 @@ manually moving files into the task workspace.
|
||||
- A successful submission makes every added source visible as a named top-level entry in the Files
|
||||
panel. Repository sources also appear in repository-aware Changes, branch, editor, and pull
|
||||
request surfaces; folder sources remain file-only.
|
||||
- Duplicate repository/branch pairs, duplicate canonical folder paths, cross-workspace repository
|
||||
- Contradictory repository/branch pairs, contradictory canonical folder paths, cross-workspace repository
|
||||
IDs, invalid remote URLs, and inaccessible local paths are rejected before the task changes.
|
||||
- A source file rename may stay within its canonical workspace/source root; a cross-root move or
|
||||
rename is rejected before either source is mutated.
|
||||
@@ -127,7 +127,7 @@ identity or make folders participate in Git operations.
|
||||
|
||||
The response returns the persisted source projection, the effective task workspace path, and the
|
||||
affected session IDs. Validation errors return `400`, ownership/not-found errors return `404`,
|
||||
duplicates or an active turn return `409`, and materialization failures return `422` after rollback.
|
||||
contradictory duplicates or an active turn return `409`, and materialization failures return `422` after rollback. Exact normalized retries succeed as no-ops.
|
||||
|
||||
The backend publishes `task.updated` with both `repositories` and `workspace_folders`, then emits a
|
||||
session-scoped workspace-sources update after agentctl has adopted the new workspace root. Clients
|
||||
@@ -147,7 +147,11 @@ outside the effective workspace remain non-actionable and are never rewritten in
|
||||
file request.
|
||||
|
||||
`add_workspace_sources_kandev` accepts the same source union and defaults `task_id` to the current
|
||||
task.
|
||||
task. The caller may target only itself or a same-workspace direct child. The task MCP server injects
|
||||
the bound caller task/session provenance and the backend verifies it before mutation; provenance is
|
||||
not an agent-supplied tool argument. Exact normalized repository and folder retries are idempotent,
|
||||
while contradictory source identities remain conflicts. See ADR
|
||||
[`2026-08-19-parent-authorized-child-workspace-sources`](../../decisions/2026-08-19-parent-authorized-child-workspace-sources.md).
|
||||
|
||||
`add_branch_to_task_kandev` preserves its existing request arguments and returns:
|
||||
|
||||
@@ -184,7 +188,7 @@ persisted in source URLs or copied into agent-visible metadata.
|
||||
| -------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| A turn or tool call is active for batch attachment | The UI disables the action when known; a racing batch request returns `409` without mutation. |
|
||||
| The invoking agent calls legacy add-branch during its turn | The worktree is created and trackers refresh without stopping the active agent, terminals, or workspace processes. |
|
||||
| Any source is invalid or duplicated | The full batch is rejected before persistence or materialization. |
|
||||
| Any source is invalid or contradicts an existing source identity | The full batch is rejected before persistence or materialization. Exact normalized retries are no-ops. |
|
||||
| A host materializer fails | New filesystem entries and source records are rolled back; existing task contents remain. |
|
||||
| A container/remote repository clone fails | Newly created remote entries are removed best-effort, durable attachments are rolled back, and the response identifies the failed source. |
|
||||
| A container/remote task submits a folder source | The request returns `422` without persistence or filesystem changes. |
|
||||
@@ -281,16 +285,18 @@ persisted; every relaunch and resume of that task reuses the persisted name.
|
||||
- **GIVEN** a live legacy add-branch materialization fails, **WHEN** the MCP call returns an error,
|
||||
**THEN** the new `task_repositories` row and any newly created repository entity are rolled back
|
||||
while the agent and existing processes continue running.
|
||||
- **GIVEN** the same repository/branch or canonical folder is already attached, **WHEN** it is
|
||||
submitted again, **THEN** the request returns a conflict naming the duplicate and leaves the task
|
||||
unchanged.
|
||||
- **GIVEN** a source identity conflicts with an already attached repository/branch or canonical folder,
|
||||
**WHEN** it is submitted, **THEN** the request returns a conflict and leaves the task unchanged.
|
||||
- **GIVEN** the same normalized repository/branch or canonical folder is already attached, **WHEN** it is
|
||||
submitted again, **THEN** the request succeeds without changing the task.
|
||||
- **GIVEN** a phone viewport on the Files tab, **WHEN** the user opens the 44px workspace-actions
|
||||
control, **THEN** an inset bottom-sheet menu exposes both actions with touch-sized rows.
|
||||
- **GIVEN** that phone action menu, **WHEN** the user selects **Add Repositories to workspace**,
|
||||
chooses repository kinds from the touch-sized **Add repository** menu, adds two sources, and
|
||||
submits, **THEN** a touch-usable full-height picker completes the same operation without
|
||||
horizontal document overflow and returns focus to the workspace-actions control.
|
||||
- **GIVEN** an agent calls `add_workspace_sources_kandev` for its current idle task, **WHEN** all
|
||||
- **GIVEN** an agent calls `add_workspace_sources_kandev` for its current idle task or an idle
|
||||
same-workspace direct child, **WHEN** all
|
||||
inputs materialize, **THEN** the UI receives the same task and session updates as the human flow.
|
||||
- **GIVEN** two distinct tasks whose titles sanitize to the same task-root slug, **WHEN** each task
|
||||
materializes a Kandev-owned task root, **THEN** their collision-resistant suffixes normally produce
|
||||
|
||||
Reference in New Issue
Block a user