feat: Introduce agent teams, agent linking, and advanced agent orchestration features.

cmd/gateway.go

@@ -640,7 +640,7 @@ func runGateway() {
 	var instanceLoader *channels.InstanceLoader
 	if managedStores != nil && managedStores.ChannelInstances != nil {
 		instanceLoader = channels.NewInstanceLoader(managedStores.ChannelInstances, managedStores.Agents, channelMgr, msgBus, pairingStore)
-		instanceLoader.RegisterFactory("telegram", telegram.FactoryWithAgentStore(managedStores.Agents))
+		instanceLoader.RegisterFactory("telegram", telegram.FactoryWithStores(managedStores.Agents, managedStores.Teams))
 		instanceLoader.RegisterFactory("discord", discord.Factory)
 		instanceLoader.RegisterFactory("feishu", feishu.Factory)
 		instanceLoader.RegisterFactory("zalo_oa", zalo.Factory)
@@ -653,7 +653,7 @@ func runGateway() {
 	// Register config-based channels as fallback (standalone mode only).
 	// In managed mode, channels are loaded from DB via instanceLoader — skip config-based registration.
 	if cfg.Channels.Telegram.Enabled && cfg.Channels.Telegram.Token != "" && instanceLoader == nil {
-		tg, err := telegram.New(cfg.Channels.Telegram, msgBus, pairingStore, nil)
+		tg, err := telegram.New(cfg.Channels.Telegram, msgBus, pairingStore, nil, nil)
 		if err != nil {
 			slog.Error("failed to initialize telegram channel", "error", err)
 		} else {
@@ -710,6 +710,16 @@ func runGateway() {
 		methods.NewChannelInstancesMethods(managedStores.ChannelInstances, msgBus).Register(server.Router())
 	}
 
+	// Register agent links WS RPC methods (managed mode only)
+	if managedStores != nil && managedStores.AgentLinks != nil && managedStores.Agents != nil {
+		methods.NewAgentLinksMethods(managedStores.AgentLinks, managedStores.Agents, agentRouter).Register(server.Router())
+	}
+
+	// Register agent teams WS RPC methods (managed mode only)
+	if managedStores != nil && managedStores.Teams != nil {
+		methods.NewTeamsMethods(managedStores.Teams, managedStores.Agents, managedStores.AgentLinks, agentRouter).Register(server.Router())
+	}
+
 	// Cache invalidation: reload channel instances on changes.
 	// Runs in a goroutine because Reload() is heavy (stops channels, waits for polling exit,
 	// sleeps 500ms, reloads from DB, starts new channels) and Broadcast handlers must be non-blocking.

cmd/gateway_agents.go

@@ -107,6 +107,7 @@ func createAgentLoop(agentID string, cfg *config.Config, router *agent.Router, p
 		ContextFiles:      contextFiles,
 		EnsureUserFiles:   ensureUserFiles,
 		ContextFileLoader: contextFileLoader,
+		BootstrapCleanup:  buildBootstrapCleanup(agentStore),
 		CompactionCfg:      cfg.Agents.Defaults.Compaction,
 		ContextPruningCfg:  cfg.Agents.Defaults.ContextPruning,
 		SandboxEnabled:         sandboxEnabled,

cmd/gateway_callbacks.go

@@ -47,6 +47,15 @@ func buildEnsureUserFiles(as store.AgentStore) agent.EnsureUserFilesFunc {
 	}
 }
 
+// buildBootstrapCleanup creates a callback that removes BOOTSTRAP.md for a user.
+// Used as a safety net after enough conversation turns, in case the LLM
+// didn't clear BOOTSTRAP.md itself. Idempotent — no-op if already cleared.
+func buildBootstrapCleanup(as store.AgentStore) agent.BootstrapCleanupFunc {
+	return func(ctx context.Context, agentID uuid.UUID, userID string) error {
+		return as.DeleteUserContextFile(ctx, agentID, userID, bootstrap.BootstrapFile)
+	}
+}
+
 // buildContextFileLoader creates the per-request context file loader callback.
 // Used by both managed and standalone modes — no mode-specific logic.
 // Delegates to the ContextFileInterceptor for type-aware routing.

cmd/gateway_consumer.go

@@ -297,7 +297,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
 			}
 
 			// Schedule through subagent lane
-			outCh := sched.Schedule(ctx, "subagent", agent.RunRequest{
+			outCh := sched.Schedule(ctx, scheduler.LaneSubagent, agent.RunRequest{
 				SessionKey:       sessionKey,
 				Message:          msg.Content,
 				Channel:          origChannel,
@@ -342,6 +342,146 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
 			continue
 		}
 
+		// --- Delegate announce: bypass debounce, inject into parent agent session ---
+		// Same pattern as subagent announce above, using "delegate" lane.
+		if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "delegate:") {
+			origChannel := msg.Metadata["origin_channel"]
+			origPeerKind := msg.Metadata["origin_peer_kind"]
+			parentAgent := msg.Metadata["parent_agent"]
+			if parentAgent == "" {
+				parentAgent = "default"
+			}
+			if origPeerKind == "" {
+				origPeerKind = string(sessions.PeerDirect)
+			}
+
+			if origChannel == "" || msg.ChatID == "" {
+				slog.Warn("delegate announce: missing origin", "sender", msg.SenderID)
+				continue
+			}
+
+			sessionKey := sessions.BuildScopedSessionKey(parentAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey)
+
+			slog.Info("delegate announce → scheduler (delegate lane)",
+				"delegation", msg.SenderID,
+				"target", msg.Metadata["target_agent"],
+				"session", sessionKey,
+			)
+
+			announceUserID := msg.UserID
+			if origPeerKind == string(sessions.PeerGroup) && msg.ChatID != "" {
+				announceUserID = fmt.Sprintf("group:%s:%s", origChannel, msg.ChatID)
+			}
+
+			// Extract parent trace context for announce linking (same as subagent announce)
+			var parentTraceID, parentRootSpanID uuid.UUID
+			if tid := msg.Metadata["origin_trace_id"]; tid != "" {
+				parentTraceID, _ = uuid.Parse(tid)
+			}
+			if sid := msg.Metadata["origin_root_span_id"]; sid != "" {
+				parentRootSpanID, _ = uuid.Parse(sid)
+			}
+
+			outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{
+				SessionKey:       sessionKey,
+				Message:          msg.Content,
+				Channel:          origChannel,
+				ChatID:           msg.ChatID,
+				PeerKind:         origPeerKind,
+				UserID:           announceUserID,
+				RunID:            fmt.Sprintf("delegate-announce-%s", msg.Metadata["delegation_id"]),
+				Stream:           false,
+				ParentTraceID:    parentTraceID,
+				ParentRootSpanID: parentRootSpanID,
+			})
+
+			go func(origCh, chatID, senderID string) {
+				outcome := <-outCh
+				if outcome.Err != nil {
+					slog.Error("delegate announce: agent run failed", "error", outcome.Err)
+					msgBus.PublishOutbound(bus.OutboundMessage{
+						Channel: origCh,
+						ChatID:  chatID,
+						Content: formatAgentError(outcome.Err),
+					})
+					return
+				}
+				if outcome.Result.Content == "" || agent.IsSilentReply(outcome.Result.Content) {
+					slog.Info("delegate announce: suppressed silent/empty reply", "delegation", senderID)
+					return
+				}
+				msgBus.PublishOutbound(bus.OutboundMessage{
+					Channel: origCh,
+					ChatID:  chatID,
+					Content: outcome.Result.Content,
+				})
+			}(origChannel, msg.ChatID, msg.SenderID)
+			continue
+		}
+
+		// --- Teammate message: bypass debounce, route to target agent session ---
+		// Same pattern as delegate announce, using "delegate" lane.
+		if msg.Channel == "system" && strings.HasPrefix(msg.SenderID, "teammate:") {
+			origChannel := msg.Metadata["origin_channel"]
+			origPeerKind := msg.Metadata["origin_peer_kind"]
+			targetAgent := msg.AgentID // team_message sets AgentID to the target agent key
+			if targetAgent == "" {
+				targetAgent = cfg.ResolveDefaultAgentID()
+			}
+			if origPeerKind == "" {
+				origPeerKind = string(sessions.PeerDirect)
+			}
+
+			if origChannel == "" || msg.ChatID == "" {
+				slog.Warn("teammate message: missing origin", "sender", msg.SenderID)
+				continue
+			}
+
+			sessionKey := sessions.BuildScopedSessionKey(targetAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey)
+
+			slog.Info("teammate message → scheduler (delegate lane)",
+				"from", msg.SenderID,
+				"to", targetAgent,
+				"session", sessionKey,
+			)
+
+			announceUserID := msg.UserID
+			if origPeerKind == string(sessions.PeerGroup) && msg.ChatID != "" {
+				announceUserID = fmt.Sprintf("group:%s:%s", origChannel, msg.ChatID)
+			}
+
+			outCh := sched.Schedule(ctx, scheduler.LaneDelegate, agent.RunRequest{
+				SessionKey: sessionKey,
+				Message:    msg.Content,
+				Channel:    origChannel,
+				ChatID:     msg.ChatID,
+				PeerKind:   origPeerKind,
+				UserID:     announceUserID,
+				RunID:      fmt.Sprintf("teammate-%s-%s", msg.Metadata["from_agent"], msg.Metadata["to_agent"]),
+				Stream:     false,
+			})
+
+			go func(origCh, chatID, senderID string) {
+				outcome := <-outCh
+				if outcome.Err != nil {
+					slog.Error("teammate message: agent run failed", "error", outcome.Err)
+					return
+				}
+				if outcome.Result.Content == "" || agent.IsSilentReply(outcome.Result.Content) {
+					slog.Info("teammate message: suppressed silent/empty reply", "from", senderID)
+					return
+				}
+				// Deliver response to origin channel (same as delegate/subagent announce).
+				// This allows the lead to respond to users after receiving teammate updates.
+				msgBus.PublishOutbound(bus.OutboundMessage{
+					Channel: origCh,
+					ChatID:  chatID,
+					Content: outcome.Result.Content,
+				})
+			}(origChannel, msg.ChatID, msg.SenderID)
+			continue
+		}
+
 		// --- Command: /stop — cancel oldest active run for this session ---
 		// --- Command: /stopall — cancel ALL active runs + drain queue ---
 		if cmd := msg.Metadata["command"]; cmd == "stop" || cmd == "stopall" {
@@ -370,6 +510,28 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents
 				cancelled = sched.CancelOneSession(sessionKey)
 				slog.Info("inbound: /stop command", "session", sessionKey, "cancelled", cancelled)
 			}
+
+			// Publish feedback so the channel can show the result.
+			var feedback string
+			if cancelled {
+				if cmd == "stopall" {
+					feedback = "All tasks stopped."
+				} else {
+					feedback = "Task stopped."
+				}
+			} else {
+				if cmd == "stopall" {
+					feedback = "No active tasks to stop."
+				} else {
+					feedback = "No active task to stop."
+				}
+			}
+			msgBus.PublishOutbound(bus.OutboundMessage{
+				Channel:  msg.Channel,
+				ChatID:   msg.ChatID,
+				Content:  feedback,
+				Metadata: msg.Metadata,
+			})
 			continue
 		}
 

cmd/gateway_managed.go

@@ -85,6 +85,7 @@ func wireManagedExtras(
 		TraceCollector:    traceCollector,
 		EnsureUserFiles:   ensureUserFiles,
 		ContextFileLoader: contextFileLoader,
+		BootstrapCleanup:  buildBootstrapCleanup(stores.Agents),
 		InjectionAction:   injectionAction,
 		MaxMessageChars:        appCfg.Gateway.MaxMessageChars,
 		CompactionCfg:          appCfg.Agents.Defaults.Compaction,
@@ -93,6 +94,8 @@ func wireManagedExtras(
 		SandboxContainerDir:    sandboxContainerDir,
 		SandboxWorkspaceAccess: sandboxWorkspaceAccess,
 		DynamicLoader:          dynamicLoader,
+		AgentLinkStore:         stores.AgentLinks,
+		TeamStore:              stores.Teams,
 		OnEvent: func(event agent.AgentEvent) {
 			msgBus.Broadcast(bus.Event{
 				Name:    protocol.EventAgent,
@@ -234,6 +237,73 @@ func wireManagedExtras(
 		})
 	}
 
+	// Register delegate tool (inter-agent delegation) if link store is available.
+	// Uses a callback to bridge tools.DelegateRunRequest → agent.RunRequest,
+	// avoiding import cycle between tools and agent packages.
+	if stores.AgentLinks != nil && stores.Agents != nil {
+		runAgentFn := func(ctx context.Context, agentKey string, req tools.DelegateRunRequest) (*tools.DelegateRunResult, error) {
+			loop, err := agentRouter.Get(agentKey)
+			if err != nil {
+				return nil, err
+			}
+			result, err := loop.Run(ctx, agent.RunRequest{
+				SessionKey:        req.SessionKey,
+				Message:           req.Message,
+				UserID:            req.UserID,
+				Channel:           req.Channel,
+				ChatID:            req.ChatID,
+				PeerKind:          req.PeerKind,
+				RunID:             req.RunID,
+				Stream:            req.Stream,
+				ExtraSystemPrompt: req.ExtraSystemPrompt,
+			})
+			if err != nil {
+				return nil, err
+			}
+			return &tools.DelegateRunResult{
+				Content:    result.Content,
+				Iterations: result.Iterations,
+			}, nil
+		}
+		delegateMgr := tools.NewDelegateManager(runAgentFn, stores.AgentLinks, stores.Agents, msgBus)
+		if stores.Teams != nil {
+			delegateMgr.SetTeamStore(stores.Teams)
+		}
+		delegateMgr.SetSessionStore(stores.Sessions)
+		toolsReg.Register(tools.NewDelegateTool(delegateMgr))
+
+		// Register delegate_search tool (hybrid FTS + semantic agent discovery)
+		var delegateEmbProvider store.EmbeddingProvider
+		if agentStore, ok := stores.Agents.(*pg.PGAgentStore); ok {
+			memCfg := appCfg.Agents.Defaults.Memory
+			if embProvider := resolveEmbeddingProvider(appCfg, memCfg); embProvider != nil {
+				agentStore.SetEmbeddingProvider(embProvider)
+				delegateEmbProvider = embProvider
+				slog.Info("managed mode: agent embeddings enabled")
+
+				// Backfill embeddings for existing agents with frontmatter
+				go func() {
+					count, err := agentStore.BackfillAgentEmbeddings(context.Background())
+					if err != nil {
+						slog.Warn("agent embeddings backfill failed", "error", err)
+					} else if count > 0 {
+						slog.Info("agent embeddings backfill complete", "updated", count)
+					}
+				}()
+			}
+		}
+		toolsReg.Register(tools.NewDelegateSearchTool(stores.AgentLinks, delegateEmbProvider))
+		slog.Info("managed mode: delegate + delegate_search tools registered")
+	}
+
+	// Register team tools (team_tasks + team_message) if team store is available.
+	if stores.Teams != nil && stores.Agents != nil {
+		teamMgr := tools.NewTeamToolManager(stores.Teams, stores.Agents, msgBus)
+		toolsReg.Register(tools.NewTeamTasksTool(teamMgr))
+		toolsReg.Register(tools.NewTeamMessageTool(teamMgr))
+		slog.Info("managed mode: team tools registered")
+	}
+
 	slog.Info("managed mode: resolver + interceptors + cache subscribers wired")
 }
 

cmd/gateway_standalone.go

@@ -65,9 +65,9 @@ func wireStandaloneExtras(
 	// Seed predefined agents' context files to disk
 	bgCtx := context.Background()
 	for _, e := range entries {
-		if e.AgentType == "predefined" {
+		if e.AgentType == store.AgentTypePredefined {
 			id := file.AgentUUID(e.Key)
-			if _, seedErr := bootstrap.SeedToStore(bgCtx, fileStore, id, "predefined"); seedErr != nil {
+			if _, seedErr := bootstrap.SeedToStore(bgCtx, fileStore, id, store.AgentTypePredefined); seedErr != nil {
 				slog.Warn("failed to seed predefined agent context files", "agent", e.Key, "error", seedErr)
 			}
 		}

cmd/onboard_managed.go

@@ -115,13 +115,13 @@ func seedManagedData(dsn string, cfg *config.Config) error {
 		AgentKey:            "default",
 		DisplayName:         "Default Agent",
 		OwnerID:             "system",
-		AgentType:           "open",
+		AgentType:           store.AgentTypeOpen,
 		Provider:            defaultProvider,
 		Model:               modelID,
 		Workspace:           workspace,
 		RestrictToWorkspace: true,
 		IsDefault:           true,
-		Status:              "active",
+		Status:              store.AgentStatusActive,
 		SubagentsConfig:     json.RawMessage(`{"maxSpawnDepth":1,"maxConcurrent":20}`),
 	}