fix(office): guard against reports_to cycles on import (#2822)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

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 {

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")
+	}
+}

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))
+	}
+}

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.

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