diff --git a/apps/backend/internal/office/config/import.go b/apps/backend/internal/office/config/import.go index bee70a84fa..a30a5e676f 100644 --- a/apps/backend/internal/office/config/import.go +++ b/apps/backend/internal/office/config/import.go @@ -115,6 +115,9 @@ func (s *ConfigService) previewProjects( func (s *ConfigService) ApplyImport( ctx context.Context, workspaceID string, bundle *ConfigBundle, ) (*ImportResult, error) { + s.importMu.Lock() + defer s.importMu.Unlock() + return s.applyImport(ctx, workspaceID, bundle, true) } @@ -153,6 +156,10 @@ func (s *ConfigService) applyAgents( ctx context.Context, wsID string, incoming []AgentConfig, result *ImportResult, allowExternalManagers bool, ) error { + if duplicateName, ok := duplicateAgentName(incoming); ok { + return fmt.Errorf("duplicate agent name %q in import bundle", duplicateName) + } + existing, err := s.repo.ListAgentInstances(ctx, wsID) if err != nil { return err @@ -194,6 +201,17 @@ func (s *ConfigService) applyAgents( return s.applyAgentReportsTo(ctx, wsID, incoming, result, allowExternalManagers) } +func duplicateAgentName(incoming []AgentConfig) (string, bool) { + seen := make(map[string]struct{}, len(incoming)) + for _, cfg := range incoming { + if _, ok := seen[cfg.Name]; ok { + return cfg.Name, true + } + seen[cfg.Name] = struct{}{} + } + return "", false +} + // applyAgentReportsTo resolves each incoming agent's reports_to name to the // target agent's ID and persists it. It runs after every agent in the bundle // has been created or updated above, so it re-lists the workspace to pick up @@ -202,8 +220,9 @@ func (s *ConfigService) applyAgents( // also reference an agent that is only in the workspace. Filesystem syncs pass // false because the snapshot is authoritative and rows absent from it are // pruned after this import. A name that resolves to nothing (dangling -// reference) or to the agent itself is recorded as a warning and left empty -// rather than failing the import. +// reference), to the agent itself, or that would create a cycle in the +// reporting hierarchy is recorded as a warning and left empty rather than +// failing the import. func (s *ConfigService) applyAgentReportsTo( ctx context.Context, wsID string, incoming []AgentConfig, result *ImportResult, allowExternalManagers bool, @@ -213,19 +232,25 @@ func (s *ConfigService) applyAgentReportsTo( return fmt.Errorf("resolve reports_to: list agents: %w", err) } byName := make(map[string]*models.AgentInstance, len(current)) + byID := make(map[string]*models.AgentInstance, len(current)) incomingNames := bundleAgentSet(incoming) for _, a := range current { if !allowExternalManagers && !incomingNames[a.Name] { continue } byName[a.Name] = a + byID[a.ID] = a + } + incomingByName := make(map[string]AgentConfig, len(incoming)) + for _, cfg := range incoming { + incomingByName[cfg.Name] = cfg } for _, cfg := range incoming { agent, ok := byName[cfg.Name] if !ok { continue } - reportsTo, warning := resolveReportsTo(cfg, byName) + reportsTo, warning := resolveReportsTo(cfg, byName, byID, incomingByName) if warning != "" { result.Warnings = append(result.Warnings, warning) } @@ -241,10 +266,17 @@ func (s *ConfigService) applyAgentReportsTo( } // resolveReportsTo resolves cfg.ReportsTo (a name) against byName (a name -> -// agent index built from the target workspace after apply). It returns the -// resolved manager ID, or an empty string plus a warning when the name is a -// self-reference or does not match any known agent. -func resolveReportsTo(cfg AgentConfig, byName map[string]*models.AgentInstance) (string, string) { +// agent index built from the target workspace after apply, already filtered +// to the callers' allowExternalManagers policy). It returns the resolved +// manager ID, or an empty string plus a warning when the name is a +// self-reference, does not match any known agent, or would create a cycle in +// the reporting hierarchy. +func resolveReportsTo( + cfg AgentConfig, + byName map[string]*models.AgentInstance, + byID map[string]*models.AgentInstance, + incomingByName map[string]AgentConfig, +) (string, string) { if cfg.ReportsTo == "" { return "", "" } @@ -255,9 +287,59 @@ func resolveReportsTo(cfg AgentConfig, byName map[string]*models.AgentInstance) if !ok { return "", fmt.Sprintf("agent %q reports_to %q, which was not found", cfg.Name, cfg.ReportsTo) } + if reportsToCycleExists(cfg.Name, cfg.ReportsTo, byName, byID, incomingByName) { + return "", fmt.Sprintf("agent %q reports_to %q, which would create a cycle", cfg.Name, cfg.ReportsTo) + } return manager.ID, "" } +// reportsToCycleExists walks the proposed manager chain upward from start, +// looking for target. Each hop is resolved against the proposed graph: when +// the current node is itself part of the incoming bundle, its parent is the +// bundle's declared reports_to name; otherwise its parent comes from +// byName/byID, the same allowExternalManagers-filtered view used to resolve +// the direct manager lookup above — an agent a sync would already reject as +// a manager (because it fell outside the filter) is equally invisible as an +// intermediate hop, since such an edge can never end up persisted anyway. +// The walk is iterative and bounded by a visited set (plus a hard hop cap as +// a second bound) so a pre-existing cyclic row already in the database +// cannot hang resolution of an unrelated agent. +func reportsToCycleExists( + target, start string, + byName map[string]*models.AgentInstance, + byID map[string]*models.AgentInstance, + incomingByName map[string]AgentConfig, +) bool { + visited := make(map[string]bool, len(byName)+len(incomingByName)) + node := start + // The visited set is the primary bound. Keep a small secondary cap in case + // a future graph representation makes the set incomplete. + maxHops := len(byName) + 1 + for i := 0; i < maxHops; i++ { + if node == target { + return true + } + if visited[node] { + return false + } + visited[node] = true + + var parent string + if incCfg, ok := incomingByName[node]; ok { + parent = incCfg.ReportsTo + } else if agent, ok := byName[node]; ok && agent.ReportsTo != "" { + if parentAgent, ok := byID[agent.ReportsTo]; ok { + parent = parentAgent.Name + } + } + if parent == "" { + return false + } + node = parent + } + return false +} + func (s *ConfigService) applySkills( ctx context.Context, wsID string, incoming []SkillConfig, result *ImportResult, ) error { diff --git a/apps/backend/internal/office/config/import_concurrency_test.go b/apps/backend/internal/office/config/import_concurrency_test.go new file mode 100644 index 0000000000..222efb8372 --- /dev/null +++ b/apps/backend/internal/office/config/import_concurrency_test.go @@ -0,0 +1,89 @@ +package config + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" +) + +type blockingActivityLogger struct { + calls atomic.Int32 + firstEntered chan struct{} + secondSeen chan struct{} + release chan struct{} + once sync.Once +} + +func (l *blockingActivityLogger) LogActivity( + _ context.Context, _, _, _, _, _, _, _ string, +) { + switch l.calls.Add(1) { + case 1: + l.once.Do(func() { close(l.firstEntered) }) + <-l.release + case 2: + close(l.secondSeen) + } +} + +func (l *blockingActivityLogger) LogActivityWithRun( + ctx context.Context, wsID, actorType, actorID, action, targetType, targetID, details, _, _ string, +) { + l.LogActivity(ctx, wsID, actorType, actorID, action, targetType, targetID, details) +} + +// TestApplyImport_SerializesConcurrentRequests prevents two imports for the +// same service from validating and writing hierarchy state at the same time. +func TestApplyImport_SerializesConcurrentRequests(t *testing.T) { + env := newTestEnv(t) + logger := &blockingActivityLogger{ + firstEntered: make(chan struct{}), + secondSeen: make(chan struct{}), + release: make(chan struct{}), + } + env.svc.activity = logger + t.Cleanup(func() { + select { + case <-logger.release: + default: + close(logger.release) + } + }) + + firstDone := make(chan error, 1) + go func() { + _, err := env.svc.ApplyImport(context.Background(), testWorkspaceID, &ConfigBundle{ + Agents: []AgentConfig{hierarchyAgent("alice", "")}, + }) + firstDone <- err + }() + <-logger.firstEntered + + secondDone := make(chan error, 1) + go func() { + _, err := env.svc.ApplyImport(context.Background(), testWorkspaceID, &ConfigBundle{ + Agents: []AgentConfig{hierarchyAgent("bob", "")}, + }) + secondDone <- err + }() + + secondEntered := false + select { + case <-logger.secondSeen: + secondEntered = true + case <-time.After(100 * time.Millisecond): + } + + close(logger.release) + if err := <-firstDone; err != nil { + t.Fatalf("first import: %v", err) + } + if err := <-secondDone; err != nil { + t.Fatalf("second import: %v", err) + } + if secondEntered { + t.Fatal("second import entered activity logging before the first import released") + } +} diff --git a/apps/backend/internal/office/config/import_reports_to_test.go b/apps/backend/internal/office/config/import_reports_to_test.go index 5673defb83..7ac28dccca 100644 --- a/apps/backend/internal/office/config/import_reports_to_test.go +++ b/apps/backend/internal/office/config/import_reports_to_test.go @@ -208,3 +208,209 @@ func TestApplyIncoming_DoesNotKeepReportsToForPrunedManager(t *testing.T) { t.Fatalf("pruned manager %q is still visible in the repository", manager.ID) } } + +// TestApplyImport_ReportsToTwoAgentCycleWarns asserts a 2-cycle formed +// entirely within one bundle (bob reports_to carol, carol reports_to bob) is +// detected, both edges are cleared, and each is reported with its own +// warning rather than silently persisted. +func TestApplyImport_ReportsToTwoAgentCycleWarns(t *testing.T) { + env := newTestEnv(t) + ctx := context.Background() + bundle := &ConfigBundle{Agents: []AgentConfig{ + hierarchyAgent("bob", "carol"), + hierarchyAgent("carol", "bob"), + }} + + result, err := env.svc.ApplyImport(ctx, testWorkspaceID, bundle) + if err != nil { + t.Fatalf("ApplyImport: %v", err) + } + if len(result.Warnings) != 2 { + t.Fatalf("warnings: got %d, want 2: %v", len(result.Warnings), result.Warnings) + } + for _, w := range result.Warnings { + if !strings.Contains(w, "bob") || !strings.Contains(w, "carol") { + t.Errorf("warning %q does not name both agents in the cycle", w) + } + } + assertEqual(t, "bob reports_to cleared", agentByName(t, env, testWorkspaceID, "bob").ReportsTo, "") + assertEqual(t, "carol reports_to cleared", agentByName(t, env, testWorkspaceID, "carol").ReportsTo, "") +} + +// TestApplyImport_ReportsToThreeAgentChainCycleWarns asserts a longer chain +// cycle (a -> b -> c -> a), not just a direct 2-cycle, is detected. Every +// edge in the cycle is cleared with its own warning. +func TestApplyImport_ReportsToThreeAgentChainCycleWarns(t *testing.T) { + env := newTestEnv(t) + ctx := context.Background() + bundle := &ConfigBundle{Agents: []AgentConfig{ + hierarchyAgent("a", "b"), + hierarchyAgent("b", "c"), + hierarchyAgent("c", "a"), + }} + + result, err := env.svc.ApplyImport(ctx, testWorkspaceID, bundle) + if err != nil { + t.Fatalf("ApplyImport: %v", err) + } + if len(result.Warnings) != 3 { + t.Fatalf("warnings: got %d, want 3: %v", len(result.Warnings), result.Warnings) + } + assertEqual(t, "a reports_to cleared", agentByName(t, env, testWorkspaceID, "a").ReportsTo, "") + assertEqual(t, "b reports_to cleared", agentByName(t, env, testWorkspaceID, "b").ReportsTo, "") + assertEqual(t, "c reports_to cleared", agentByName(t, env, testWorkspaceID, "c").ReportsTo, "") +} + +// TestApplyImport_ReportsToCycleThroughPersistedNonBundleAgentWarns asserts +// a cycle is still detected when only one side of it is in the incoming +// bundle: frank already exists in the workspace, eve already persistently +// reports_to frank (outside this bundle), and this bundle proposes frank +// reports_to eve. Resolving only against the bundle's own contents would +// miss this; the walk must also consult persisted state. +func TestApplyImport_ReportsToCycleThroughPersistedNonBundleAgentWarns(t *testing.T) { + env := newTestEnv(t) + ctx := context.Background() + + seed := &ConfigBundle{Agents: []AgentConfig{hierarchyAgent("frank", "")}} + if _, err := env.svc.ApplyImport(ctx, testWorkspaceID, seed); err != nil { + t.Fatalf("seed frank: %v", err) + } + frank := agentByName(t, env, testWorkspaceID, "frank") + + eve := seedAgent(t, env, testWorkspaceID, "eve") + eve.ReportsTo = frank.ID + if err := env.repo.UpdateAgentInstance(ctx, eve); err != nil { + t.Fatalf("seed eve reports_to frank: %v", err) + } + + bundle := &ConfigBundle{Agents: []AgentConfig{hierarchyAgent("frank", "eve")}} + result, err := env.svc.ApplyImport(ctx, testWorkspaceID, bundle) + if err != nil { + t.Fatalf("ApplyImport: %v", err) + } + if len(result.Warnings) != 1 { + t.Fatalf("warnings: got %d, want 1: %v", len(result.Warnings), result.Warnings) + } + if !strings.Contains(result.Warnings[0], "frank") || !strings.Contains(result.Warnings[0], "eve") { + t.Errorf("warning %q does not name both agents in the cycle", result.Warnings[0]) + } + assertEqual(t, "frank reports_to cleared", agentByName(t, env, testWorkspaceID, "frank").ReportsTo, "") + assertEqual(t, "eve reports_to unchanged", agentByName(t, env, testWorkspaceID, "eve").ReportsTo, frank.ID) +} + +// TestApplyImport_ReportsToDeepChainNoCycleResolves guards against +// over-rejection: a valid deep chain with no cycle (a -> b -> c) must still +// resolve every edge with zero warnings. +func TestApplyImport_ReportsToDeepChainNoCycleResolves(t *testing.T) { + env := newTestEnv(t) + ctx := context.Background() + bundle := &ConfigBundle{Agents: []AgentConfig{ + hierarchyAgent("a", "b"), + hierarchyAgent("b", "c"), + hierarchyAgent("c", ""), + }} + + result, err := env.svc.ApplyImport(ctx, testWorkspaceID, bundle) + if err != nil { + t.Fatalf("ApplyImport: %v", err) + } + if len(result.Warnings) != 0 { + t.Fatalf("warnings: got %d, want 0: %v", len(result.Warnings), result.Warnings) + } + a := agentByName(t, env, testWorkspaceID, "a") + b := agentByName(t, env, testWorkspaceID, "b") + c := agentByName(t, env, testWorkspaceID, "c") + assertEqual(t, "a reports_to b", a.ReportsTo, b.ID) + assertEqual(t, "b reports_to c", b.ReportsTo, c.ID) + assertEqual(t, "c reports_to", c.ReportsTo, "") +} + +// TestApplyImport_PreExistingPersistedCycleDoesNotPreventResolution asserts +// that a cyclic reports_to pair already sitting in the database (from before this +// guard existed) cannot hang import resolution for an unrelated agent. The +// walk must be bounded by a visited set, not rely on the graph being +// acyclic. +func TestApplyImport_PreExistingPersistedCycleDoesNotPreventResolution(t *testing.T) { + env := newTestEnv(t) + ctx := context.Background() + + x := seedAgent(t, env, testWorkspaceID, "x") + y := seedAgent(t, env, testWorkspaceID, "y") + x.ReportsTo = y.ID + if err := env.repo.UpdateAgentInstance(ctx, x); err != nil { + t.Fatalf("seed x reports_to y: %v", err) + } + y.ReportsTo = x.ID + if err := env.repo.UpdateAgentInstance(ctx, y); err != nil { + t.Fatalf("seed y reports_to x: %v", err) + } + + bundle := &ConfigBundle{Agents: []AgentConfig{hierarchyAgent("zack", "x")}} + if _, err := env.svc.ApplyImport(ctx, testWorkspaceID, bundle); err != nil { + t.Fatalf("ApplyImport: %v", err) + } + + zack := agentByName(t, env, testWorkspaceID, "zack") + assertEqual(t, "zack reports_to x", zack.ReportsTo, x.ID) +} + +// TestApplyIncoming_CycleThroughPrunedExternalManagerDoesNotFalseFlag +// asserts the filesystem-sync path (allowExternalManagers=false) does not +// walk into a pruned, DB-only agent's persisted reports_to when checking a +// bundle member for a cycle. grace is DB-only (absent from the filesystem) +// and persistently reports_to bob; the filesystem snapshot proposes bob +// reports_to carol and carol reports_to grace. bob must resolve cleanly, while +// carol must receive the expected dangling-reference warning. +func TestApplyIncoming_CycleThroughPrunedExternalManagerDoesNotFalseFlag(t *testing.T) { + env := newTestEnv(t) + ctx := context.Background() + + grace := seedAgent(t, env, testWorkspaceID, "grace") + bob := seedAgent(t, env, testWorkspaceID, "bob") + grace.ReportsTo = bob.ID + if err := env.repo.UpdateAgentInstance(ctx, grace); err != nil { + t.Fatalf("seed grace reports_to bob: %v", err) + } + + env.writeFSAgent(t, "bob", "name: bob\nrole: worker\nreports_to: carol\n") + env.writeFSAgent(t, "carol", "name: carol\nrole: manager\nreports_to: grace\n") + + result, err := env.svc.ApplyIncoming(ctx, testWorkspaceID) + if err != nil { + t.Fatalf("ApplyIncoming: %v", err) + } + if len(result.Warnings) != 1 { + t.Fatalf("warnings: got %d, want 1: %v", len(result.Warnings), result.Warnings) + } + if !strings.Contains(result.Warnings[0], "carol") || !strings.Contains(result.Warnings[0], "grace") { + t.Errorf("warning %q does not name the pruned manager and report", result.Warnings[0]) + } + carol := agentByName(t, env, testWorkspaceID, "carol") + assertEqual(t, "bob reports_to carol after fs sync", agentByName(t, env, testWorkspaceID, "bob").ReportsTo, carol.ID) + assertEqual(t, "carol reports_to cleared", carol.ReportsTo, "") + if _, err := env.repo.GetAgentInstance(ctx, grace.ID); err == nil { + t.Fatalf("pruned manager %q is still visible in the repository", grace.ID) + } +} + +// TestApplyImport_DuplicateAgentNamesFailBeforeWrites rejects an ambiguous +// bundle before it can create duplicate rows or build an incorrect graph. +func TestApplyImport_DuplicateAgentNamesFailBeforeWrites(t *testing.T) { + env := newTestEnv(t) + ctx := context.Background() + bundle := &ConfigBundle{Agents: []AgentConfig{ + hierarchyAgent("bob", "carol"), + hierarchyAgent("bob", "dave"), + }} + + if _, err := env.svc.ApplyImport(ctx, testWorkspaceID, bundle); err == nil { + t.Fatal("ApplyImport succeeded for a bundle with duplicate agent names") + } + agents, err := env.repo.ListAgentInstances(ctx, testWorkspaceID) + if err != nil { + t.Fatalf("list agents: %v", err) + } + if len(agents) != 0 { + t.Fatalf("duplicate bundle partially wrote %d agent rows", len(agents)) + } +} diff --git a/apps/backend/internal/office/config/service.go b/apps/backend/internal/office/config/service.go index 19fb7eae6e..e193674976 100644 --- a/apps/backend/internal/office/config/service.go +++ b/apps/backend/internal/office/config/service.go @@ -6,6 +6,7 @@ import ( "context" "fmt" "io" + "sync" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/office/configloader" @@ -26,6 +27,8 @@ type ConfigService struct { cfgWriter *configloader.FileWriter logger *logger.Logger activity shared.ActivityLogger + // importMu keeps each import's read/validate/write lifecycle atomic. + importMu sync.Mutex } // NewConfigService constructs a ConfigService. diff --git a/apps/backend/internal/office/config/sync.go b/apps/backend/internal/office/config/sync.go index 8980338168..e125d2e804 100644 --- a/apps/backend/internal/office/config/sync.go +++ b/apps/backend/internal/office/config/sync.go @@ -93,6 +93,9 @@ func (s *ConfigService) OutgoingDiff(ctx context.Context, workspaceID string) (* // ApplyIncoming reads the on-disk config and writes it to the DB. Rows in the // DB but missing from disk are deleted. func (s *ConfigService) ApplyIncoming(ctx context.Context, workspaceID string) (*ImportResult, error) { + s.importMu.Lock() + defer s.importMu.Unlock() + bundle, _, err := s.ScanFilesystem(ctx, workspaceID) if err != nil { return nil, err