diff --git a/cmd/gateway.go b/cmd/gateway.go index d4e16ad0..9f61bac3 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -747,24 +747,24 @@ func runGateway() { // 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()) + methods.NewAgentLinksMethods(managedStores.AgentLinks, managedStores.Agents, agentRouter, msgBus).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()) + methods.NewTeamsMethods(managedStores.Teams, managedStores.Agents, managedStores.AgentLinks, agentRouter, msgBus).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. if instanceLoader != nil { - msgBus.Subscribe("cache:channel_instances", func(event bus.Event) { + msgBus.Subscribe(bus.TopicCacheChannelInstances, func(event bus.Event) { if event.Name != protocol.EventCacheInvalidate { return } payload, ok := event.Payload.(bus.CacheInvalidatePayload) - if !ok || payload.Kind != "channel_instances" { + if !ok || payload.Kind != bus.CacheKindChannelInstances { return } go instanceLoader.Reload(context.Background()) @@ -843,7 +843,7 @@ func runGateway() { // Subscribe to agent events for channel streaming/reaction forwarding. // Events emitted by agent loops are broadcast to the bus; we forward them // to the channel manager which routes to StreamingChannel/ReactionChannel. - msgBus.Subscribe("channel-streaming", func(event bus.Event) { + msgBus.Subscribe(bus.TopicChannelStreaming, func(event bus.Event) { if event.Name != protocol.EventAgent { return } diff --git a/cmd/gateway_callbacks.go b/cmd/gateway_callbacks.go index 5437d8b0..87277077 100644 --- a/cmd/gateway_callbacks.go +++ b/cmd/gateway_callbacks.go @@ -17,13 +17,13 @@ import ( // Used by both managed and standalone modes — no mode-specific logic. // Seeds per-user context files on first chat (new user profile). func buildEnsureUserFiles(as store.AgentStore) agent.EnsureUserFilesFunc { - return func(ctx context.Context, agentID uuid.UUID, userID, agentType, workspace string) error { - isNew, err := as.GetOrCreateUserProfile(ctx, agentID, userID, workspace) + return func(ctx context.Context, agentID uuid.UUID, userID, agentType, workspace, channel string) (string, error) { + isNew, effectiveWs, err := as.GetOrCreateUserProfile(ctx, agentID, userID, workspace, channel) if err != nil { - return err + return effectiveWs, err } if !isNew { - return nil // already profiled = already seeded + return effectiveWs, nil // already profiled = already seeded } // Auto-add first group member as a file writer (bootstrap the allowlist). @@ -43,7 +43,7 @@ func buildEnsureUserFiles(as store.AgentStore) agent.EnsureUserFilesFunc { } _, err = bootstrap.SeedUserFiles(ctx, as, agentID, userID, agentType) - return err + return effectiveWs, err } } diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index 709a64e0..a0d3dcc7 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -162,7 +162,7 @@ func wireManagedExtras( // Context file cache: invalidate on agent/context data changes if contextFileInterceptor != nil { - msgBus.Subscribe("cache:bootstrap", func(event bus.Event) { + msgBus.Subscribe(bus.TopicCacheBootstrap, func(event bus.Event) { if event.Name != protocol.EventCacheInvalidate { return } @@ -170,7 +170,7 @@ func wireManagedExtras( if !ok { return } - if payload.Kind == "bootstrap" || payload.Kind == "agent" { + if payload.Kind == bus.CacheKindBootstrap || payload.Kind == bus.CacheKindAgent { if payload.Key != "" { agentID, err := uuid.Parse(payload.Key) if err == nil { @@ -184,12 +184,12 @@ func wireManagedExtras( } // Agent router: invalidate Loop cache on agent config changes - msgBus.Subscribe("cache:agent", func(event bus.Event) { + msgBus.Subscribe(bus.TopicCacheAgent, func(event bus.Event) { if event.Name != protocol.EventCacheInvalidate { return } payload, ok := event.Payload.(bus.CacheInvalidatePayload) - if !ok || payload.Kind != "agent" { + if !ok || payload.Kind != bus.CacheKindAgent { return } if payload.Key != "" { @@ -199,12 +199,12 @@ func wireManagedExtras( // Skills cache: bump version on skill changes if stores.Skills != nil { - msgBus.Subscribe("cache:skills", func(event bus.Event) { + msgBus.Subscribe(bus.TopicCacheSkills, func(event bus.Event) { if event.Name != protocol.EventCacheInvalidate { return } payload, ok := event.Payload.(bus.CacheInvalidatePayload) - if !ok || payload.Kind != "skills" { + if !ok || payload.Kind != bus.CacheKindSkills { return } stores.Skills.BumpVersion() @@ -213,12 +213,12 @@ func wireManagedExtras( // Cron cache: invalidate job cache on cron changes if ci, ok := stores.Cron.(store.CacheInvalidatable); ok { - msgBus.Subscribe("cache:cron", func(event bus.Event) { + msgBus.Subscribe(bus.TopicCacheCron, func(event bus.Event) { if event.Name != protocol.EventCacheInvalidate { return } payload, ok := event.Payload.(bus.CacheInvalidatePayload) - if !ok || payload.Kind != "cron" { + if !ok || payload.Kind != bus.CacheKindCron { return } ci.InvalidateCache() @@ -227,12 +227,12 @@ func wireManagedExtras( // Custom tools cache: reload global tools on create/update/delete if dynamicLoader != nil { - msgBus.Subscribe("cache:custom_tools", func(event bus.Event) { + msgBus.Subscribe(bus.TopicCacheCustomTools, func(event bus.Event) { if event.Name != protocol.EventCacheInvalidate { return } payload, ok := event.Payload.(bus.CacheInvalidatePayload) - if !ok || payload.Kind != "custom_tools" { + if !ok || payload.Kind != bus.CacheKindCustomTools { return } dynamicLoader.ReloadGlobal(context.Background(), toolsReg) @@ -243,7 +243,7 @@ func wireManagedExtras( // Builtin tools cache: re-apply disables on settings/enabled changes if stores.BuiltinTools != nil { - msgBus.Subscribe("cache:builtin_tools", func(event bus.Event) { + msgBus.Subscribe(bus.TopicCacheBuiltinTools, func(event bus.Event) { if event.Name != protocol.EventCacheInvalidate { return } @@ -280,8 +280,9 @@ func wireManagedExtras( return nil, err } dr := &tools.DelegateRunResult{ - Content: result.Content, - Iterations: result.Iterations, + Content: result.Content, + Iterations: result.Iterations, + Deliverables: result.Deliverables, } for _, m := range result.Media { dr.MediaPaths = append(dr.MediaPaths, m.Path) @@ -352,9 +353,35 @@ func wireManagedExtras( teamMgr := tools.NewTeamToolManager(stores.Teams, stores.Agents, msgBus) toolsReg.Register(tools.NewTeamTasksTool(teamMgr)) toolsReg.Register(tools.NewTeamMessageTool(teamMgr)) + + // Team cache invalidation via pub/sub + msgBus.Subscribe(bus.TopicCacheTeam, func(event bus.Event) { + if event.Name != protocol.EventCacheInvalidate { + return + } + payload, ok := event.Payload.(bus.CacheInvalidatePayload) + if !ok || payload.Kind != bus.CacheKindTeam { + return + } + teamMgr.InvalidateTeam() + }) slog.Info("managed mode: team tools registered") } + // User workspace cache: invalidate per-user workspace path on profile changes + msgBus.Subscribe(bus.TopicCacheUserWorkspace, func(event bus.Event) { + if event.Name != protocol.EventCacheInvalidate { + return + } + payload, ok := event.Payload.(bus.CacheInvalidatePayload) + if !ok || payload.Kind != bus.CacheKindUserWorkspace { + return + } + if payload.Key != "" { + agentRouter.InvalidateUserWorkspace(payload.Key) + } + }) + slog.Info("managed mode: resolver + interceptors + cache subscribers wired") return contextFileInterceptor } diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 4845151f..9a234266 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -32,7 +32,8 @@ import ( const bootstrapAutoCleanupTurns = 3 // EnsureUserFilesFunc seeds per-user context files on first chat (managed mode). -type EnsureUserFilesFunc func(ctx context.Context, agentID uuid.UUID, userID, agentType, workspace string) error +// Returns the effective workspace path (from user_agent_profiles) for caching. +type EnsureUserFilesFunc func(ctx context.Context, agentID uuid.UUID, userID, agentType, workspace, channel string) (effectiveWorkspace string, err error) // ContextFileLoaderFunc loads context files dynamically per-request (managed mode). type ContextFileLoaderFunc func(ctx context.Context, agentID uuid.UUID, userID, agentType string) []bootstrap.ContextFile @@ -74,7 +75,7 @@ type Loop struct { ensureUserFiles EnsureUserFilesFunc contextFileLoader ContextFileLoaderFunc bootstrapCleanup BootstrapCleanupFunc - seededUsers sync.Map // userID → true, avoid re-check per request + userWorkspaces sync.Map // userID → string (expanded workspace path from user_agent_profiles) // Compaction config (memory flush settings) compactionCfg *config.CompactionConfig @@ -253,11 +254,12 @@ type RunRequest struct { // RunResult is the output of a completed agent run. type RunResult struct { - Content string `json:"content"` - RunID string `json:"runId"` - Iterations int `json:"iterations"` - Usage *providers.Usage `json:"usage,omitempty"` - Media []MediaResult `json:"media,omitempty"` // media files from tool results (MEDIA: prefix) + Content string `json:"content"` + RunID string `json:"runId"` + Iterations int `json:"iterations"` + Usage *providers.Usage `json:"usage,omitempty"` + Media []MediaResult `json:"media,omitempty"` // media files from tool results (MEDIA: prefix) + Deliverables []string `json:"deliverables,omitempty"` // actual content from tool outputs (for team task results) } // MediaResult represents a media file produced by a tool during the agent run. @@ -397,25 +399,37 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } // Per-user workspace isolation. - // Each user gets a subdirectory within the agent's workspace. - if l.workspace != "" { - effectiveWorkspace := l.workspace - if req.UserID != "" { - effectiveWorkspace = filepath.Join(l.workspace, sanitizePathSegment(req.UserID)) - if err := os.MkdirAll(effectiveWorkspace, 0755); err != nil { - slog.Warn("failed to create user workspace directory", "workspace", effectiveWorkspace, "user", req.UserID, "error", err) + // Workspace path comes from user_agent_profiles (includes channel segment + // for cross-channel isolation). Cached in userWorkspaces to avoid repeated DB queries. + if l.workspace != "" && req.UserID != "" { + cachedWs, loaded := l.userWorkspaces.Load(req.UserID) + if !loaded { + // First request for this user: get/create profile → returns stored workspace. + // Also seeds per-user context files on first chat (managed mode). + ws := l.workspace + if l.ensureUserFiles != nil { + var err error + ws, err = l.ensureUserFiles(ctx, l.agentUUID, req.UserID, l.agentType, l.workspace, req.Channel) + if err != nil { + slog.Warn("failed to ensure user context files", "error", err) + ws = l.workspace + } } + // Expand ~ and convert to absolute for filesystem operations. + ws = config.ExpandHome(ws) + if !filepath.IsAbs(ws) { + ws, _ = filepath.Abs(ws) + } + l.userWorkspaces.Store(req.UserID, ws) + cachedWs = ws + } + effectiveWorkspace := filepath.Join(cachedWs.(string), sanitizePathSegment(req.UserID)) + if err := os.MkdirAll(effectiveWorkspace, 0755); err != nil { + slog.Warn("failed to create user workspace directory", "workspace", effectiveWorkspace, "user", req.UserID, "error", err) } ctx = tools.WithToolWorkspace(ctx, effectiveWorkspace) - } - - // Ensure per-user context files exist (first-chat seeding, managed mode) - if l.ensureUserFiles != nil && req.UserID != "" { - if _, loaded := l.seededUsers.LoadOrStore(req.UserID, true); !loaded { - if err := l.ensureUserFiles(ctx, l.agentUUID, req.UserID, l.agentType, l.workspace); err != nil { - slog.Warn("failed to ensure user context files", "error", err) - } - } + } else if l.workspace != "" { + ctx = tools.WithToolWorkspace(ctx, l.workspace) } // Persist agent UUID + user ID on the session (for querying/tracing) @@ -513,8 +527,9 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) var totalUsage providers.Usage iteration := 0 var finalContent string - var asyncToolCalls []string // track async spawn tool names for fallback + var asyncToolCalls []string // track async spawn tool names for fallback var mediaResults []MediaResult // media files from tool MEDIA: results + var deliverables []string // actual content from tool outputs (for team task results) // Inject retry hook so channels can update placeholder on LLM retries. ctx = providers.WithRetryHook(ctx, func(attempt, maxAttempts int, err error) { @@ -674,6 +689,9 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) for _, p := range result.Media { mediaResults = append(mediaResults, MediaResult{Path: p, ContentType: mimeFromExt(filepath.Ext(p))}) } + if result.Deliverable != "" { + deliverables = append(deliverables, result.Deliverable) + } toolMsg := providers.Message{ Role: "tool", @@ -785,6 +803,9 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) for _, p := range r.result.Media { mediaResults = append(mediaResults, MediaResult{Path: p, ContentType: mimeFromExt(filepath.Ext(p))}) } + if r.result.Deliverable != "" { + deliverables = append(deliverables, r.result.Deliverable) + } toolMsg := providers.Message{ Role: "tool", @@ -890,11 +911,12 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } return &RunResult{ - Content: finalContent, - RunID: req.RunID, - Iterations: iteration, - Usage: &totalUsage, - Media: mediaResults, + Content: finalContent, + RunID: req.RunID, + Iterations: iteration, + Usage: &totalUsage, + Media: mediaResults, + Deliverables: deliverables, }, nil } @@ -988,3 +1010,9 @@ func sanitizePathSegment(s string) string { } return b.String() } + +// InvalidateUserWorkspace clears the cached workspace for a user, +// forcing the next request to re-read from user_agent_profiles. +func (l *Loop) InvalidateUserWorkspace(userID string) { + l.userWorkspaces.Delete(userID) +} diff --git a/internal/agent/loop_history.go b/internal/agent/loop_history.go index e85d47cf..d10fc5b9 100644 --- a/internal/agent/loop_history.go +++ b/internal/agent/loop_history.go @@ -28,10 +28,15 @@ func (l *Loop) buildMessages(ctx context.Context, history []providers.Message, s _, hasSpawn := l.tools.Get("spawn") _, hasSkillSearch := l.tools.Get("skill_search") - // Per-user workspace: show the user's subdirectory in the system prompt (managed mode) + // Per-user workspace: show the user's subdirectory in the system prompt (managed mode). + // Uses cached workspace from user_agent_profiles (includes channel isolation). promptWorkspace := l.workspace if l.agentUUID != uuid.Nil && userID != "" && l.workspace != "" { - promptWorkspace = filepath.Join(l.workspace, sanitizePathSegment(userID)) + if cachedWs, ok := l.userWorkspaces.Load(userID); ok { + promptWorkspace = filepath.Join(cachedWs.(string), sanitizePathSegment(userID)) + } else { + promptWorkspace = filepath.Join(l.workspace, sanitizePathSegment(userID)) + } } // Resolve context files once — also detect BOOTSTRAP.md presence. diff --git a/internal/agent/resolver.go b/internal/agent/resolver.go index 5d5779a2..b33fdc5b 100644 --- a/internal/agent/resolver.go +++ b/internal/agent/resolver.go @@ -340,6 +340,9 @@ func buildDelegateAgentsMD(targets []store.AgentLinkData) string { if t.TargetDisplayName != "" { sb.WriteString(fmt.Sprintf(" (%s)", t.TargetDisplayName)) } + if t.TargetIsTeamLead && t.TargetTeamName != "" { + sb.WriteString(fmt.Sprintf(" [Team Lead: %s]", t.TargetTeamName)) + } sb.WriteString("\n") if t.TargetDescription != "" { sb.WriteString(t.TargetDescription + "\n") diff --git a/internal/agent/router.go b/internal/agent/router.go index 7615b462..0521550e 100644 --- a/internal/agent/router.go +++ b/internal/agent/router.go @@ -177,6 +177,18 @@ func (r *Router) AbortRun(runID, sessionKey string) bool { return true } +// InvalidateUserWorkspace clears the cached workspace for a user across all cached agent loops. +// Used when user_agent_profiles.workspace changes (e.g. admin reassignment). +func (r *Router) InvalidateUserWorkspace(userID string) { + r.mu.RLock() + defer r.mu.RUnlock() + for _, entry := range r.agents { + if loop, ok := entry.agent.(*Loop); ok { + loop.InvalidateUserWorkspace(userID) + } + } +} + // AbortRunsForSession cancels all active runs for a session key. // Returns the list of aborted run IDs. func (r *Router) AbortRunsForSession(sessionKey string) []string { diff --git a/internal/bootstrap/seed_store_test.go b/internal/bootstrap/seed_store_test.go index 533beab2..c02f46b7 100644 --- a/internal/bootstrap/seed_store_test.go +++ b/internal/bootstrap/seed_store_test.go @@ -79,8 +79,8 @@ func (s *seedStubStore) GetUserOverride(_ context.Context, _ uuid.UUID, _ string func (s *seedStubStore) SetUserOverride(_ context.Context, _ *store.UserAgentOverrideData) error { return nil } -func (s *seedStubStore) GetOrCreateUserProfile(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) { - return false, nil +func (s *seedStubStore) GetOrCreateUserProfile(_ context.Context, _ uuid.UUID, _, _, _ string) (bool, string, error) { + return false, "", nil } func (s *seedStubStore) IsGroupFileWriter(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) { return false, nil diff --git a/internal/bootstrap/templates/SOUL.md b/internal/bootstrap/templates/SOUL.md index 792306ac..d1fa1d1e 100644 --- a/internal/bootstrap/templates/SOUL.md +++ b/internal/bootstrap/templates/SOUL.md @@ -25,6 +25,10 @@ _You're not a chatbot. You're becoming someone._ Be the assistant you'd actually want to talk to. Concise when needed, thorough when it matters. Not a corporate drone. Not a sycophant. Just... good. +## Expertise + +_(Optional — add domain-specific knowledge, technical skills, or specialized instructions here. Remove this placeholder when customizing.)_ + ## Continuity Each session, you wake up fresh. These files _are_ your memory. Read them. Update them. They're how you persist. diff --git a/internal/bus/types.go b/internal/bus/types.go index 6d9c5baf..c0d2483c 100644 --- a/internal/bus/types.go +++ b/internal/bus/types.go @@ -48,6 +48,22 @@ const ( CacheKindCustomTools = "custom_tools" CacheKindChannelInstances = "channel_instances" CacheKindBuiltinTools = "builtin_tools" + CacheKindTeam = "team" + CacheKindUserWorkspace = "user_workspace" +) + +// Topic constants for msgBus.Subscribe() / Broadcast(). +const ( + TopicCacheBootstrap = "cache:bootstrap" + TopicCacheAgent = "cache:agent" + TopicCacheSkills = "cache:skills" + TopicCacheCron = "cache:cron" + TopicCacheCustomTools = "cache:custom_tools" + TopicCacheBuiltinTools = "cache:builtin_tools" + TopicCacheTeam = "cache:team" + TopicCacheUserWorkspace = "cache:user_workspace" + TopicCacheChannelInstances = "cache:channel_instances" + TopicChannelStreaming = "channel-streaming" ) // CacheInvalidatePayload signals cache layers to evict stale entries. diff --git a/internal/channels/telegram/send.go b/internal/channels/telegram/send.go index 6f94d33a..45d53a81 100644 --- a/internal/channels/telegram/send.go +++ b/internal/channels/telegram/send.go @@ -241,10 +241,11 @@ func (c *Channel) sendMediaMessage(ctx context.Context, chatID int64, msg bus.Ou // Only reply to the first media item replyTo = 0 - // Send follow-up text if caption was split + // Send follow-up text if caption was split. + // followUpText is already HTML (from markdownToTelegramHTML above), so + // just chunk and send — do NOT convert again (double-escaping breaks entities). if followUpText != "" { - htmlContent := markdownToTelegramHTML(followUpText) - chunks := chunkHTML(htmlContent, telegramMaxMessageLen) + chunks := chunkHTML(followUpText, telegramMaxMessageLen) for _, chunk := range chunks { if err := c.sendHTML(ctx, chatID, chunk, 0, threadID); err != nil { return err diff --git a/internal/config/config_load.go b/internal/config/config_load.go index b2a2d0e7..7fc132cc 100644 --- a/internal/config/config_load.go +++ b/internal/config/config_load.go @@ -368,3 +368,16 @@ func ExpandHome(path string) string { } return home } + +// ContractHome replaces the user home directory prefix with ~. +// Reverse of ExpandHome — used to store portable paths in the database. +func ContractHome(path string) string { + if path == "" { + return path + } + home, _ := os.UserHomeDir() + if home != "" && strings.HasPrefix(path, home) { + return "~" + path[len(home):] + } + return path +} diff --git a/internal/gateway/methods/agent_links.go b/internal/gateway/methods/agent_links.go index ea0676e4..92e819f7 100644 --- a/internal/gateway/methods/agent_links.go +++ b/internal/gateway/methods/agent_links.go @@ -7,6 +7,7 @@ import ( "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/agent" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" @@ -14,13 +15,14 @@ import ( // AgentLinksMethods handles agents.links.* RPC methods. type AgentLinksMethods struct { - linkStore store.AgentLinkStore - agentStore store.AgentStore - agentRouter *agent.Router // for cache invalidation when links change + linkStore store.AgentLinkStore + agentStore store.AgentStore + agentRouter *agent.Router // for cache invalidation when links change + msgBus *bus.MessageBus // for pub/sub cache invalidation } -func NewAgentLinksMethods(linkStore store.AgentLinkStore, agentStore store.AgentStore, agentRouter *agent.Router) *AgentLinksMethods { - return &AgentLinksMethods{linkStore: linkStore, agentStore: agentStore, agentRouter: agentRouter} +func NewAgentLinksMethods(linkStore store.AgentLinkStore, agentStore store.AgentStore, agentRouter *agent.Router, msgBus *bus.MessageBus) *AgentLinksMethods { + return &AgentLinksMethods{linkStore: linkStore, agentStore: agentStore, agentRouter: agentRouter, msgBus: msgBus} } func (m *AgentLinksMethods) Register(router *gateway.MethodRouter) { @@ -177,6 +179,7 @@ func (m *AgentLinksMethods) handleCreate(_ context.Context, client *gateway.Clie m.agentRouter.InvalidateAgent(sourceAgent.AgentKey) m.agentRouter.InvalidateAgent(targetAgent.AgentKey) } + m.emitTeamCacheInvalidate() client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ "link": link, @@ -255,6 +258,7 @@ func (m *AgentLinksMethods) handleUpdate(_ context.Context, client *gateway.Clie // Invalidate affected agents so AGENTS.md gets regenerated m.invalidateLinkAgents(ctx, linkID) + m.emitTeamCacheInvalidate() client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) } @@ -302,10 +306,23 @@ func (m *AgentLinksMethods) handleDelete(_ context.Context, client *gateway.Clie if link != nil { m.invalidateLinkAgentsByID(ctx, link.SourceAgentID, link.TargetAgentID) } + m.emitTeamCacheInvalidate() client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) } +// emitTeamCacheInvalidate broadcasts a cache invalidation event for team data. +// Called when links change since team-member links affect team resolution. +func (m *AgentLinksMethods) emitTeamCacheInvalidate() { + if m.msgBus == nil { + return + } + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventCacheInvalidate, + Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindTeam}, + }) +} + // invalidateLinkAgents fetches a link by ID and invalidates both source and target agent caches. func (m *AgentLinksMethods) invalidateLinkAgents(ctx context.Context, linkID uuid.UUID) { if m.agentRouter == nil { diff --git a/internal/gateway/methods/agents_create_owner_test.go b/internal/gateway/methods/agents_create_owner_test.go index 7b85c706..e10d6326 100644 --- a/internal/gateway/methods/agents_create_owner_test.go +++ b/internal/gateway/methods/agents_create_owner_test.go @@ -77,8 +77,8 @@ func (s *createCaptureStore) GetUserOverride(_ context.Context, _ uuid.UUID, _ s func (s *createCaptureStore) SetUserOverride(_ context.Context, _ *store.UserAgentOverrideData) error { return nil } -func (s *createCaptureStore) GetOrCreateUserProfile(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) { - return false, nil +func (s *createCaptureStore) GetOrCreateUserProfile(_ context.Context, _ uuid.UUID, _, _, _ string) (bool, string, error) { + return false, "", nil } func (s *createCaptureStore) IsGroupFileWriter(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) { return false, nil diff --git a/internal/gateway/methods/teams.go b/internal/gateway/methods/teams.go index a6f9a3ec..6451bbe5 100644 --- a/internal/gateway/methods/teams.go +++ b/internal/gateway/methods/teams.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/agent" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" @@ -19,10 +20,22 @@ type TeamsMethods struct { agentStore store.AgentStore linkStore store.AgentLinkStore // for auto-creating bidirectional links agentRouter *agent.Router // for cache invalidation + msgBus *bus.MessageBus // for pub/sub cache invalidation } -func NewTeamsMethods(teamStore store.TeamStore, agentStore store.AgentStore, linkStore store.AgentLinkStore, agentRouter *agent.Router) *TeamsMethods { - return &TeamsMethods{teamStore: teamStore, agentStore: agentStore, linkStore: linkStore, agentRouter: agentRouter} +func NewTeamsMethods(teamStore store.TeamStore, agentStore store.AgentStore, linkStore store.AgentLinkStore, agentRouter *agent.Router, msgBus *bus.MessageBus) *TeamsMethods { + return &TeamsMethods{teamStore: teamStore, agentStore: agentStore, linkStore: linkStore, agentRouter: agentRouter, msgBus: msgBus} +} + +// emitTeamCacheInvalidate broadcasts a cache invalidation event for team data. +func (m *TeamsMethods) emitTeamCacheInvalidate() { + if m.msgBus == nil { + return + } + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventCacheInvalidate, + Payload: bus.CacheInvalidatePayload{Kind: bus.CacheKindTeam}, + }) } func (m *TeamsMethods) Register(router *gateway.MethodRouter) { @@ -33,6 +46,8 @@ func (m *TeamsMethods) Register(router *gateway.MethodRouter) { router.Register(protocol.MethodTeamsTaskList, m.handleTaskList) router.Register(protocol.MethodTeamsMembersAdd, m.handleAddMember) router.Register(protocol.MethodTeamsMembersRemove, m.handleRemoveMember) + router.Register(protocol.MethodTeamsUpdate, m.handleUpdate) + router.Register(protocol.MethodTeamsKnownUsers, m.handleKnownUsers) } // --- List --- @@ -438,20 +453,125 @@ func (m *TeamsMethods) handleRemoveMember(_ context.Context, client *gateway.Cli client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) } -// invalidateTeamCaches invalidates agent caches for all members of a team. +// invalidateTeamCaches invalidates agent caches for all members of a team +// and emits a pub/sub event for TeamToolManager cache invalidation. func (m *TeamsMethods) invalidateTeamCaches(ctx context.Context, teamID uuid.UUID) { - if m.agentRouter == nil { - return - } - members, err := m.teamStore.ListMembers(ctx, teamID) - if err != nil { - return - } - for _, member := range members { - if member.AgentKey != "" { - m.agentRouter.InvalidateAgent(member.AgentKey) + if m.agentRouter != nil { + members, err := m.teamStore.ListMembers(ctx, teamID) + if err == nil { + for _, member := range members { + if member.AgentKey != "" { + m.agentRouter.InvalidateAgent(member.AgentKey) + } + } } } + m.emitTeamCacheInvalidate() +} + +// --- Update (settings) --- + +type teamsUpdateParams struct { + TeamID string `json:"teamId"` + Settings map[string]interface{} `json:"settings"` +} + +func (m *TeamsMethods) handleUpdate(_ context.Context, client *gateway.Client, req *protocol.RequestFrame) { + if m.teamStore == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "teams not available (standalone mode)")) + return + } + + var params teamsUpdateParams + if err := json.Unmarshal(req.Params, ¶ms); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid params")) + return + } + + if params.TeamID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "teamId is required")) + return + } + + teamID, err := uuid.Parse(params.TeamID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid teamId")) + return + } + + ctx := context.Background() + + // Validate team exists + if _, err := m.teamStore.GetTeam(ctx, teamID); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "team not found: "+err.Error())) + return + } + + // Validate settings against teamAccessSettings schema (strip unknown fields) + type teamAccessSettings struct { + AllowUserIDs []string `json:"allow_user_ids"` + DenyUserIDs []string `json:"deny_user_ids"` + AllowChannels []string `json:"allow_channels"` + DenyChannels []string `json:"deny_channels"` + } + raw, _ := json.Marshal(params.Settings) + var access teamAccessSettings + if err := json.Unmarshal(raw, &access); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid settings: "+err.Error())) + return + } + cleaned, _ := json.Marshal(access) + + updates := map[string]any{"settings": json.RawMessage(cleaned)} + if err := m.teamStore.UpdateTeam(ctx, teamID, updates); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "failed to update team: "+err.Error())) + return + } + + m.invalidateTeamCaches(ctx, teamID) + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) +} + +// --- Known Users --- + +type teamsKnownUsersParams struct { + TeamID string `json:"teamId"` +} + +func (m *TeamsMethods) handleKnownUsers(_ context.Context, client *gateway.Client, req *protocol.RequestFrame) { + if m.teamStore == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "teams not available (standalone mode)")) + return + } + + var params teamsKnownUsersParams + if err := json.Unmarshal(req.Params, ¶ms); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid params")) + return + } + + if params.TeamID == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "teamId is required")) + return + } + + teamID, err := uuid.Parse(params.TeamID) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, "invalid teamId")) + return + } + + ctx := context.Background() + users, err := m.teamStore.KnownUserIDs(ctx, teamID, 100) + if err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, err.Error())) + return + } + + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ + "users": users, + })) } // --- helpers --- diff --git a/internal/http/summoner.go b/internal/http/summoner.go index 9536fce0..24202a14 100644 --- a/internal/http/summoner.go +++ b/internal/http/summoner.go @@ -281,13 +281,13 @@ func (s *AgentSummoner) buildCreatePrompt(description string) string { 1. Language: Write ALL content in the SAME LANGUAGE as the . If description is in Vietnamese, write in Vietnamese. If in English, write in English. BUT keep ALL headings and section titles in English exactly as in the templates. -2. SOUL.md rules: - - KEEP the exact English headings: "# SOUL.md - Who You Are", "## Core Truths", "## Boundaries", "## Vibe", "## Continuity" - - KEEP the general advice in "## Core Truths" — do NOT inject agent-specific references there. Core Truths are universal personality traits. - - CUSTOMIZE "## Vibe" to reflect this agent's unique personality and communication style. - - CUSTOMIZE "## Boundaries" if the agent has specific boundaries mentioned in the description. - - Keep "## Continuity" as-is (just translate if needed). - - Do NOT add the agent's name or role references into Core Truths or Boundaries. +2. SOUL.md section guide — each section has a specific purpose: + - "## Core Truths" — universal personality traits. KEEP the general advice. Do NOT inject agent-specific references here. + - "## Boundaries" — rules and limits. CUSTOMIZE only if the description mentions specific boundaries. + - "## Vibe" — communication style and personality ONLY. How the agent talks, its tone, its attitude. Do NOT put technical knowledge here. + - "## Expertise" — domain-specific knowledge, technical skills, specialized instructions, keywords, parameters. If the description mentions any specialized domain (e.g. image generation, coding, writing), put that knowledge HERE. Remove the placeholder text. If no domain expertise, omit this section entirely. + - "## Continuity" — keep as-is (just translate if needed). + - KEEP the exact English headings. Do NOT add the agent's name into Core Truths or Boundaries. 3. IDENTITY.md rules: - KEEP the exact English heading: "# IDENTITY.md - Who Am I?" @@ -333,18 +333,34 @@ func (s *AgentSummoner) buildEditPrompt(existing []store.AgentContextFileData, e sb.WriteString("\n\n") fmt.Fprintf(&sb, "\n%s\n\n\n", editPrompt) sb.WriteString(`IMPORTANT RULES: -1. Write ALL content in the SAME LANGUAGE as the existing files. Only keep headings and technical terms in English. -2. Keep ALL English headings exactly as they are (e.g. "## Core Truths", "## Vibe"). -3. Only output files that need changes. Omit unchanged files. + +1. Language: Write ALL content in the SAME LANGUAGE as the existing files. Keep headings in English. + +2. SOUL.md section guide — place content in the RIGHT section: + - "## Core Truths" — universal personality traits. Do NOT add domain-specific content here. + - "## Boundaries" — rules and limits. + - "## Vibe" — communication style and personality ONLY. Tone, attitude, how the agent talks. NOT technical knowledge. + - "## Expertise" — domain-specific knowledge, technical skills, keywords, parameters, specialized instructions. If the edit adds domain knowledge (e.g. image generation techniques, coding standards, writing styles), it goes HERE. Create this section if it doesn't exist yet (between Vibe and Continuity). + - "## Continuity" — memory/persistence rules. Usually unchanged. + +3. Output the COMPLETE updated file content, not just the changed parts. The output will REPLACE the entire file. + +4. Only output files that actually need changes. Omit unchanged files entirely. + +5. If the edit changes the agent's expertise, also update the frontmatter summary. Output format: + +(updated expertise summary, or omit if unchanged) + + -(updated content, or omit if unchanged) +(complete updated content) -(updated content, or omit if unchanged) +(complete updated content, or omit if unchanged) `) return sb.String() diff --git a/internal/sessions/manager.go b/internal/sessions/manager.go index 8b447f95..90035164 100644 --- a/internal/sessions/manager.go +++ b/internal/sessions/manager.go @@ -326,6 +326,8 @@ func (m *Manager) List(agentID string) []SessionInfo { MessageCount: len(s.Messages), Created: s.Created, Updated: s.Updated, + Label: s.Label, + Channel: s.Channel, }) } return result @@ -375,6 +377,8 @@ type SessionInfo struct { MessageCount int `json:"messageCount"` Created time.Time `json:"created"` Updated time.Time `json:"updated"` + Label string `json:"label,omitempty"` + Channel string `json:"channel,omitempty"` } // Save persists a session to disk atomically. diff --git a/internal/store/agent_link_store.go b/internal/store/agent_link_store.go index 7bc4cb73..8b67b8de 100644 --- a/internal/store/agent_link_store.go +++ b/internal/store/agent_link_store.go @@ -38,7 +38,9 @@ type AgentLinkData struct { TargetAgentKey string `json:"target_agent_key,omitempty"` TargetDisplayName string `json:"target_display_name,omitempty"` TargetDescription string `json:"target_description,omitempty"` - TeamName string `json:"team_name,omitempty"` // from LEFT JOIN agent_teams + TeamName string `json:"team_name,omitempty"` // from LEFT JOIN agent_teams (link's own team) + TargetIsTeamLead bool `json:"target_is_team_lead,omitempty"` // true if target is lead of any active team + TargetTeamName string `json:"target_team_name,omitempty"` // name of team the target leads } // AgentLinkStore manages inter-agent delegation links. diff --git a/internal/store/agent_store.go b/internal/store/agent_store.go index 56bb74a3..914432ac 100644 --- a/internal/store/agent_store.go +++ b/internal/store/agent_store.go @@ -197,7 +197,7 @@ type AgentStore interface { SetUserOverride(ctx context.Context, override *UserAgentOverrideData) error // User-agent profiles - GetOrCreateUserProfile(ctx context.Context, agentID uuid.UUID, userID, workspace string) (isNew bool, err error) + GetOrCreateUserProfile(ctx context.Context, agentID uuid.UUID, userID, workspace, channel string) (isNew bool, effectiveWorkspace string, err error) // Group file writers (allowlist for protected file edits in group chats) IsGroupFileWriter(ctx context.Context, agentID uuid.UUID, groupID, userID string) (bool, error) diff --git a/internal/store/file/agents.go b/internal/store/file/agents.go index e2297797..58c456e0 100644 --- a/internal/store/file/agents.go +++ b/internal/store/file/agents.go @@ -261,28 +261,39 @@ func (s *FileAgentStore) DeleteUserContextFile(_ context.Context, agentID uuid.U // --- User profiles (SQLite) --- -func (s *FileAgentStore) GetOrCreateUserProfile(_ context.Context, agentID uuid.UUID, userID, workspace string) (bool, error) { +func (s *FileAgentStore) GetOrCreateUserProfile(_ context.Context, agentID uuid.UUID, userID, workspace, channel string) (bool, string, error) { + // Build workspace with channel segment for isolation. + effectiveWs := workspace + if channel != "" { + effectiveWs = filepath.Join(workspace, channel) + } + result, err := s.db.Exec( `INSERT OR IGNORE INTO user_profiles (agent_id, user_id, workspace) VALUES (?, ?, ?)`, - agentID.String(), userID, workspace, + agentID.String(), userID, effectiveWs, ) if err != nil { - return false, err + return false, effectiveWs, err } affected, err := result.RowsAffected() if err != nil { - return false, err + return false, effectiveWs, err } if affected > 0 { - return true, nil // new profile + return true, effectiveWs, nil // new profile } - // Existing profile — update last_seen - _, _ = s.db.Exec( - `UPDATE user_profiles SET last_seen_at = CURRENT_TIMESTAMP WHERE agent_id = ? AND user_id = ?`, + // Existing profile — update last_seen, return stored workspace + var storedWs sql.NullString + _ = s.db.QueryRow( + `UPDATE user_profiles SET last_seen_at = CURRENT_TIMESTAMP WHERE agent_id = ? AND user_id = ? RETURNING workspace`, agentID.String(), userID, - ) - return false, nil + ).Scan(&storedWs) + ws := effectiveWs + if storedWs.Valid && storedWs.String != "" { + ws = storedWs.String + } + return false, ws, nil } // --- Group file writers (SQLite) --- diff --git a/internal/store/file/sessions.go b/internal/store/file/sessions.go index 01747315..eb977df1 100644 --- a/internal/store/file/sessions.go +++ b/internal/store/file/sessions.go @@ -111,6 +111,8 @@ func (f *FileSessionStore) List(agentID string) []store.SessionInfo { MessageCount: item.MessageCount, Created: item.Created, Updated: item.Updated, + Label: item.Label, + Channel: item.Channel, } } return result diff --git a/internal/store/pg/agent_links.go b/internal/store/pg/agent_links.go index f5753bb8..971c817f 100644 --- a/internal/store/pg/agent_links.go +++ b/internal/store/pg/agent_links.go @@ -28,6 +28,20 @@ const linkSelectCols = `id, source_agent_id, target_agent_id, direction, team_id const linkSelectColsJoined = `l.id, l.source_agent_id, l.target_agent_id, l.direction, l.team_id, l.description, l.max_concurrent, l.settings, l.status, l.created_by, l.created_at, l.updated_at` +// targetTeamLeadCols detects if the "target" agent (the other side of the link) is a team lead. +// Uses $1 = fromAgentID to determine which side is the target via CASE. +const targetTeamLeadCols = `EXISTS( + SELECT 1 FROM agent_teams tl + WHERE tl.lead_agent_id = CASE WHEN l.source_agent_id = $1 THEN l.target_agent_id ELSE l.source_agent_id END + AND tl.status = 'active' + ) AS target_is_team_lead, + COALESCE(( + SELECT tl.name FROM agent_teams tl + WHERE tl.lead_agent_id = CASE WHEN l.source_agent_id = $1 THEN l.target_agent_id ELSE l.source_agent_id END + AND tl.status = 'active' + LIMIT 1 + ), '') AS target_team_name` + func (s *PGAgentLinkStore) CreateLink(ctx context.Context, link *store.AgentLinkData) error { if link.ID == uuid.Nil { link.ID = store.GenNewID() @@ -77,7 +91,9 @@ func (s *PGAgentLinkStore) ListLinksFrom(ctx context.Context, agentID uuid.UUID) ta.agent_key AS target_agent_key, COALESCE(ta.display_name, '') AS target_display_name, COALESCE(ta.frontmatter, '') AS target_description, - COALESCE(tm.name, '') AS team_name + COALESCE(tm.name, '') AS team_name, + EXISTS(SELECT 1 FROM agent_teams tl WHERE tl.lead_agent_id = l.target_agent_id AND tl.status = 'active') AS target_is_team_lead, + COALESCE((SELECT tl.name FROM agent_teams tl WHERE tl.lead_agent_id = l.target_agent_id AND tl.status = 'active' LIMIT 1), '') AS target_team_name FROM agent_links l JOIN agents sa ON sa.id = l.source_agent_id JOIN agents ta ON ta.id = l.target_agent_id @@ -98,7 +114,9 @@ func (s *PGAgentLinkStore) ListLinksTo(ctx context.Context, agentID uuid.UUID) ( ta.agent_key AS target_agent_key, COALESCE(ta.display_name, '') AS target_display_name, COALESCE(ta.frontmatter, '') AS target_description, - COALESCE(tm.name, '') AS team_name + COALESCE(tm.name, '') AS team_name, + EXISTS(SELECT 1 FROM agent_teams tl WHERE tl.lead_agent_id = l.target_agent_id AND tl.status = 'active') AS target_is_team_lead, + COALESCE((SELECT tl.name FROM agent_teams tl WHERE tl.lead_agent_id = l.target_agent_id AND tl.status = 'active' LIMIT 1), '') AS target_team_name FROM agent_links l JOIN agents sa ON sa.id = l.source_agent_id JOIN agents ta ON ta.id = l.target_agent_id @@ -134,7 +152,8 @@ func (s *PGAgentLinkStore) DelegateTargets(ctx context.Context, fromAgentID uuid CASE WHEN l.source_agent_id = $1 THEN ta.agent_key ELSE sa.agent_key END AS target_agent_key, CASE WHEN l.source_agent_id = $1 THEN COALESCE(ta.display_name, '') ELSE COALESCE(sa.display_name, '') END AS target_display_name, CASE WHEN l.source_agent_id = $1 THEN COALESCE(ta.frontmatter, '') ELSE COALESCE(sa.frontmatter, '') END AS target_description, - COALESCE(tm.name, '') AS team_name + COALESCE(tm.name, '') AS team_name, + `+targetTeamLeadCols+` FROM agent_links l JOIN agents sa ON sa.id = l.source_agent_id JOIN agents ta ON ta.id = l.target_agent_id @@ -179,7 +198,8 @@ func (s *PGAgentLinkStore) SearchDelegateTargets(ctx context.Context, fromAgentI CASE WHEN l.source_agent_id = $1 THEN ta.agent_key ELSE sa.agent_key END AS target_agent_key, CASE WHEN l.source_agent_id = $1 THEN COALESCE(ta.display_name, '') ELSE COALESCE(sa.display_name, '') END AS target_display_name, CASE WHEN l.source_agent_id = $1 THEN COALESCE(ta.frontmatter, '') ELSE COALESCE(sa.frontmatter, '') END AS target_description, - COALESCE(tm.name, '') AS team_name + COALESCE(tm.name, '') AS team_name, + `+targetTeamLeadCols+` FROM agent_links l JOIN agents sa ON sa.id = l.source_agent_id JOIN agents ta ON ta.id = l.target_agent_id @@ -211,7 +231,8 @@ func (s *PGAgentLinkStore) SearchDelegateTargetsByEmbedding(ctx context.Context, CASE WHEN l.source_agent_id = $1 THEN ta.agent_key ELSE sa.agent_key END AS target_agent_key, CASE WHEN l.source_agent_id = $1 THEN COALESCE(ta.display_name, '') ELSE COALESCE(sa.display_name, '') END AS target_display_name, CASE WHEN l.source_agent_id = $1 THEN COALESCE(ta.frontmatter, '') ELSE COALESCE(sa.frontmatter, '') END AS target_description, - COALESCE(tm.name, '') AS team_name + COALESCE(tm.name, '') AS team_name, + `+targetTeamLeadCols+` FROM agent_links l JOIN agents sa ON sa.id = l.source_agent_id JOIN agents ta ON ta.id = l.target_agent_id @@ -267,7 +288,7 @@ func scanLinkRowsJoined(rows *sql.Rows) ([]store.AgentLinkData, error) { &d.ID, &d.SourceAgentID, &d.TargetAgentID, &d.Direction, &d.TeamID, &desc, &d.MaxConcurrent, &d.Settings, &d.Status, &d.CreatedBy, &d.CreatedAt, &d.UpdatedAt, &d.SourceAgentKey, &d.TargetAgentKey, &d.TargetDisplayName, &d.TargetDescription, - &d.TeamName, + &d.TeamName, &d.TargetIsTeamLead, &d.TargetTeamName, ); err != nil { return nil, err } diff --git a/internal/store/pg/agents_context.go b/internal/store/pg/agents_context.go index 5b00b0e5..f9d5e0ee 100644 --- a/internal/store/pg/agents_context.go +++ b/internal/store/pg/agents_context.go @@ -3,10 +3,12 @@ package pg import ( "context" "database/sql" + "path/filepath" "time" "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/store" ) @@ -81,15 +83,30 @@ func (s *PGAgentStore) DeleteUserContextFile(ctx context.Context, agentID uuid.U // --- User-Agent Profiles --- -func (s *PGAgentStore) GetOrCreateUserProfile(ctx context.Context, agentID uuid.UUID, userID, workspace string) (bool, error) { +func (s *PGAgentStore) GetOrCreateUserProfile(ctx context.Context, agentID uuid.UUID, userID, workspace, channel string) (bool, string, error) { + // Build workspace with channel segment for isolation. + // Store in portable ~ form (e.g. "~/.goclaw/agent-ws/telegram"). + effectiveWs := config.ContractHome(workspace) + if channel != "" { + effectiveWs = filepath.Join(effectiveWs, channel) + } + var isInserted bool + var storedWorkspace sql.NullString err := s.db.QueryRowContext(ctx, ` INSERT INTO user_agent_profiles (agent_id, user_id, workspace, first_seen_at, last_seen_at) VALUES ($1, $2, NULLIF($3, ''), NOW(), NOW()) - ON CONFLICT (agent_id, user_id) DO UPDATE SET last_seen_at = NOW(), workspace = EXCLUDED.workspace - RETURNING (xmax = 0) - `, agentID, userID, workspace).Scan(&isInserted) - return isInserted, err + ON CONFLICT (agent_id, user_id) DO UPDATE SET last_seen_at = NOW() + RETURNING (xmax = 0), workspace + `, agentID, userID, effectiveWs).Scan(&isInserted, &storedWorkspace) + if err != nil { + return false, effectiveWs, err + } + ws := effectiveWs + if storedWorkspace.Valid && storedWorkspace.String != "" { + ws = storedWorkspace.String + } + return isInserted, ws, nil } // --- User Overrides --- diff --git a/internal/store/pg/sessions.go b/internal/store/pg/sessions.go index 40f90595..ea135f85 100644 --- a/internal/store/pg/sessions.go +++ b/internal/store/pg/sessions.go @@ -283,10 +283,10 @@ func (s *PGSessionStore) List(agentID string) []store.SessionInfo { if agentID != "" { prefix := "agent:" + agentID + ":%" rows, err = s.db.Query( - "SELECT session_key, messages, created_at, updated_at FROM sessions WHERE session_key LIKE $1 ORDER BY updated_at DESC", prefix) + "SELECT session_key, messages, created_at, updated_at, label, channel FROM sessions WHERE session_key LIKE $1 ORDER BY updated_at DESC", prefix) } else { rows, err = s.db.Query( - "SELECT session_key, messages, created_at, updated_at FROM sessions ORDER BY updated_at DESC") + "SELECT session_key, messages, created_at, updated_at, label, channel FROM sessions ORDER BY updated_at DESC") } if err != nil { return nil @@ -298,7 +298,8 @@ func (s *PGSessionStore) List(agentID string) []store.SessionInfo { var key string var msgsJSON []byte var createdAt, updatedAt time.Time - if err := rows.Scan(&key, &msgsJSON, &createdAt, &updatedAt); err != nil { + var label, channel *string + if err := rows.Scan(&key, &msgsJSON, &createdAt, &updatedAt, &label, &channel); err != nil { continue } var msgs []providers.Message @@ -308,6 +309,8 @@ func (s *PGSessionStore) List(agentID string) []store.SessionInfo { MessageCount: len(msgs), Created: createdAt, Updated: updatedAt, + Label: derefStr(label), + Channel: derefStr(channel), }) } return result @@ -343,11 +346,11 @@ func (s *PGSessionStore) ListPaged(opts store.SessionListOpts) store.SessionList var selectArgs []interface{} if opts.AgentID != "" { - selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at + selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel FROM sessions WHERE session_key LIKE $1 ORDER BY updated_at DESC LIMIT $2 OFFSET $3` selectArgs = []interface{}{whereArgs[0], limit, offset} } else { - selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at + selectQ = `SELECT session_key, jsonb_array_length(messages), created_at, updated_at, label, channel FROM sessions ORDER BY updated_at DESC LIMIT $1 OFFSET $2` selectArgs = []interface{}{limit, offset} } @@ -363,7 +366,8 @@ func (s *PGSessionStore) ListPaged(opts store.SessionListOpts) store.SessionList var key string var msgCount int var createdAt, updatedAt time.Time - if err := rows.Scan(&key, &msgCount, &createdAt, &updatedAt); err != nil { + var label, channel *string + if err := rows.Scan(&key, &msgCount, &createdAt, &updatedAt, &label, &channel); err != nil { continue } result = append(result, store.SessionInfo{ @@ -371,6 +375,8 @@ func (s *PGSessionStore) ListPaged(opts store.SessionListOpts) store.SessionList MessageCount: msgCount, Created: createdAt, Updated: updatedAt, + Label: derefStr(label), + Channel: derefStr(channel), }) } if result == nil { diff --git a/internal/store/pg/teams.go b/internal/store/pg/teams.go index 307faa91..32d72a49 100644 --- a/internal/store/pg/teams.go +++ b/internal/store/pg/teams.go @@ -60,6 +60,10 @@ func (s *PGTeamStore) GetTeam(ctx context.Context, teamID uuid.UUID) (*store.Tea return scanTeamRow(row) } +func (s *PGTeamStore) UpdateTeam(ctx context.Context, teamID uuid.UUID, updates map[string]any) error { + return execMapUpdate(ctx, s.db, "agent_teams", teamID, updates) +} + func (s *PGTeamStore) DeleteTeam(ctx context.Context, teamID uuid.UUID) error { _, err := s.db.ExecContext(ctx, `DELETE FROM agent_teams WHERE id = $1`, teamID) return err @@ -162,6 +166,33 @@ func (s *PGTeamStore) GetTeamForAgent(ctx context.Context, agentID uuid.UUID) (* return d, err } +func (s *PGTeamStore) KnownUserIDs(ctx context.Context, teamID uuid.UUID, limit int) ([]string, error) { + if limit <= 0 { + limit = 100 + } + rows, err := s.db.QueryContext(ctx, + `SELECT DISTINCT s.user_id + FROM sessions s + JOIN agent_team_members m ON m.agent_id = s.agent_id + WHERE m.team_id = $1 AND s.user_id != '' + ORDER BY s.user_id + LIMIT $2`, teamID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var users []string + for rows.Next() { + var uid string + if err := rows.Scan(&uid); err != nil { + return nil, err + } + users = append(users, uid) + } + return users, rows.Err() +} + // ============================================================ // Handoff routing // ============================================================ diff --git a/internal/store/pg/teams_tasks.go b/internal/store/pg/teams_tasks.go index 47b6729d..e8631a07 100644 --- a/internal/store/pg/teams_tasks.go +++ b/internal/store/pg/teams_tasks.go @@ -110,12 +110,12 @@ func (s *PGTeamStore) SearchTasks(ctx context.Context, teamID uuid.UUID, query s return scanTaskRowsJoined(rows) } -func (s *PGTeamStore) ClaimTask(ctx context.Context, taskID, agentID uuid.UUID) error { +func (s *PGTeamStore) ClaimTask(ctx context.Context, taskID, agentID, teamID uuid.UUID) error { res, err := s.db.ExecContext(ctx, `UPDATE team_tasks SET status = $1, owner_agent_id = $2, updated_at = $3 - WHERE id = $4 AND status = $5 AND owner_agent_id IS NULL`, + WHERE id = $4 AND status = $5 AND owner_agent_id IS NULL AND team_id = $6`, store.TeamTaskStatusInProgress, agentID, time.Now(), - taskID, store.TeamTaskStatusPending, + taskID, store.TeamTaskStatusPending, teamID, ) if err != nil { return err @@ -130,7 +130,7 @@ func (s *PGTeamStore) ClaimTask(ctx context.Context, taskID, agentID uuid.UUID) return nil } -func (s *PGTeamStore) CompleteTask(ctx context.Context, taskID uuid.UUID, result string) error { +func (s *PGTeamStore) CompleteTask(ctx context.Context, taskID, teamID uuid.UUID, result string) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err @@ -140,9 +140,9 @@ func (s *PGTeamStore) CompleteTask(ctx context.Context, taskID uuid.UUID, result // Mark task as completed (must be in_progress — use ClaimTask first) res, err := tx.ExecContext(ctx, `UPDATE team_tasks SET status = $1, result = $2, updated_at = $3 - WHERE id = $4 AND status = $5`, + WHERE id = $4 AND status = $5 AND team_id = $6`, store.TeamTaskStatusCompleted, result, time.Now(), - taskID, store.TeamTaskStatusInProgress, + taskID, store.TeamTaskStatusInProgress, teamID, ) if err != nil { return err diff --git a/internal/store/session_store.go b/internal/store/session_store.go index 1bb88a74..ff2244ff 100644 --- a/internal/store/session_store.go +++ b/internal/store/session_store.go @@ -42,6 +42,8 @@ type SessionInfo struct { MessageCount int `json:"messageCount"` Created time.Time `json:"created"` Updated time.Time `json:"updated"` + Label string `json:"label,omitempty"` + Channel string `json:"channel,omitempty"` } // SessionListOpts holds pagination options for ListPaged. diff --git a/internal/store/team_store.go b/internal/store/team_store.go index a54b7494..14fb119d 100644 --- a/internal/store/team_store.go +++ b/internal/store/team_store.go @@ -156,6 +156,7 @@ type TeamStore interface { // Team CRUD CreateTeam(ctx context.Context, team *TeamData) error GetTeam(ctx context.Context, teamID uuid.UUID) (*TeamData, error) + UpdateTeam(ctx context.Context, teamID uuid.UUID, updates map[string]any) error DeleteTeam(ctx context.Context, teamID uuid.UUID) error ListTeams(ctx context.Context) ([]TeamData, error) @@ -168,6 +169,10 @@ type TeamStore interface { // Returns nil, nil if the agent is not in any team. GetTeamForAgent(ctx context.Context, agentID uuid.UUID) (*TeamData, error) + // KnownUserIDs returns distinct user IDs from sessions of team member agents. + // Used by team settings UI to populate user select boxes. + KnownUserIDs(ctx context.Context, teamID uuid.UUID, limit int) ([]string, error) + // Tasks (shared task list) CreateTask(ctx context.Context, task *TeamTaskData) error UpdateTask(ctx context.Context, taskID uuid.UUID, updates map[string]any) error @@ -181,10 +186,12 @@ type TeamStore interface { // ClaimTask atomically transitions a task from pending to in_progress. // Only one agent can claim a given task (row-level lock, race-safe). - ClaimTask(ctx context.Context, taskID, agentID uuid.UUID) error + // teamID is validated in the WHERE clause to prevent cross-team task claiming. + ClaimTask(ctx context.Context, taskID, agentID, teamID uuid.UUID) error // CompleteTask marks a task as completed and unblocks dependent tasks. - CompleteTask(ctx context.Context, taskID uuid.UUID, result string) error + // teamID is validated in the WHERE clause to prevent cross-team task completion. + CompleteTask(ctx context.Context, taskID, teamID uuid.UUID, result string) error // Delegation history SaveDelegationHistory(ctx context.Context, record *DelegationHistoryData) error diff --git a/internal/tools/context_file_interceptor_test.go b/internal/tools/context_file_interceptor_test.go index d445a173..31dc4494 100644 --- a/internal/tools/context_file_interceptor_test.go +++ b/internal/tools/context_file_interceptor_test.go @@ -65,8 +65,8 @@ func (s *stubAgentStore) GetUserOverride(_ context.Context, _ uuid.UUID, _ strin func (s *stubAgentStore) SetUserOverride(_ context.Context, _ *store.UserAgentOverrideData) error { return nil } -func (s *stubAgentStore) GetOrCreateUserProfile(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) { - return false, nil +func (s *stubAgentStore) GetOrCreateUserProfile(_ context.Context, _ uuid.UUID, _, _, _ string) (bool, string, error) { + return false, "", nil } func (s *stubAgentStore) IsGroupFileWriter(_ context.Context, _ uuid.UUID, _, _ string) (bool, error) { return false, nil diff --git a/internal/tools/create_image.go b/internal/tools/create_image.go index c32fce1c..3ca8c520 100644 --- a/internal/tools/create_image.go +++ b/internal/tools/create_image.go @@ -115,13 +115,22 @@ func (t *CreateImageTool) Execute(ctx context.Context, args map[string]interface return ErrorResult(fmt.Sprintf("image generation failed: %v", err)) } - // Save to temp file - imagePath := filepath.Join(os.TempDir(), fmt.Sprintf("goclaw_gen_%d.png", time.Now().UnixNano())) + // Save to workspace under date-based folder (e.g. generated/2026-03-02/) + workspace := ToolWorkspaceFromCtx(ctx) + if workspace == "" { + workspace = os.TempDir() + } + dateDir := filepath.Join(workspace, "generated", time.Now().Format("2006-01-02")) + if err := os.MkdirAll(dateDir, 0755); err != nil { + return ErrorResult(fmt.Sprintf("failed to create output directory: %v", err)) + } + imagePath := filepath.Join(dateDir, fmt.Sprintf("goclaw_gen_%d.png", time.Now().UnixNano())) if err := os.WriteFile(imagePath, imageBytes, 0644); err != nil { return ErrorResult(fmt.Sprintf("failed to save generated image: %v", err)) } result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s", imagePath)} + result.Deliverable = fmt.Sprintf("[Generated image: %s]\nPrompt: %s", filepath.Base(imagePath), prompt) result.Provider = providerName result.Model = model if usage != nil { diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index a7e3c10b..532dc689 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -73,9 +73,10 @@ type DelegateRunRequest struct { // DelegateRunResult is the result from AgentRunFunc. type DelegateRunResult struct { - Content string - Iterations int - MediaPaths []string // media file paths from tool results (e.g. generated images) + Content string + Iterations int + MediaPaths []string // media file paths from tool results (e.g. generated images) + Deliverables []string // actual content from tool outputs (e.g. written file text, image prompt) } // DelegateArtifacts holds forwarded artifacts from delegation results. @@ -89,9 +90,10 @@ type DelegateArtifacts struct { // DelegateResultSummary is a compact representation of a delegation result // included in the final announce so the lead has all results in one message. type DelegateResultSummary struct { - AgentKey string - Content string - HasMedia bool + AgentKey string + Content string + HasMedia bool + Deliverables []string // actual content from tool outputs } // AgentRunFunc runs an agent by key with the given request. @@ -206,7 +208,7 @@ func (dm *DelegateManager) Delegate(ctx context.Context, opts DelegateOpts) (*De task.Status = "completed" dm.emitEvent("delegation.completed", task) dm.trackCompleted(task) - dm.autoCompleteTeamTask(task, result.Content) + dm.autoCompleteTeamTask(task, result.Content, result.Deliverables) dm.saveDelegationHistory(task, result.Content, nil, duration) slog.Info("delegation completed", "id", task.ID, "target", opts.TargetAgentKey, "iterations", result.Iterations) @@ -255,8 +257,6 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) siblingCount++ } } - isLastDelegation := siblingCount == 0 - // Announce result to parent via message bus if dm.msgBus != nil && task.OriginChannel != "" { elapsed := time.Since(task.CreatedAt) @@ -269,9 +269,10 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) if result != nil { arts.Media = result.MediaPaths arts.Results = []DelegateResultSummary{{ - AgentKey: task.TargetAgentKey, - Content: result.Content, - HasMedia: len(result.MediaPaths) > 0, + AgentKey: task.TargetAgentKey, + Content: result.Content, + HasMedia: len(result.MediaPaths) > 0, + Deliverables: result.Deliverables, }} } else if runErr != nil { arts.Results = []DelegateResultSummary{{ @@ -288,9 +289,10 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) if result != nil { artifacts.Media = append(artifacts.Media, result.MediaPaths...) artifacts.Results = append(artifacts.Results, DelegateResultSummary{ - AgentKey: task.TargetAgentKey, - Content: result.Content, - HasMedia: len(result.MediaPaths) > 0, + AgentKey: task.TargetAgentKey, + Content: result.Content, + HasMedia: len(result.MediaPaths) > 0, + Deliverables: result.Deliverables, }) } @@ -330,12 +332,15 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) dm.emitEvent("delegation.completed", task) dm.trackCompleted(task) resultContent := "" + var deliverables []string if result != nil { resultContent = result.Content - if isLastDelegation { - dm.autoCompleteTeamTask(task, resultContent) - } + deliverables = result.Deliverables } + // Auto-complete the team task for EVERY delegation (not just the last one). + // Each delegation has its own TeamTaskID — the isLastDelegation guard + // is for announce batching only, not for task completion. + dm.autoCompleteTeamTask(task, resultContent, deliverables) dm.saveDelegationHistory(task, resultContent, nil, duration) } } @@ -387,6 +392,34 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO } } + // Validate that team_task_id belongs to the agent's team (prevent cross-team task completion). + if dm.teamStore != nil && opts.TeamTaskID != uuid.Nil { + teamTask, err := dm.teamStore.GetTask(ctx, opts.TeamTaskID) + if err != nil { + return nil, nil, fmt.Errorf("team_task_id not found: %w", err) + } + if team, _ := dm.teamStore.GetTeamForAgent(ctx, sourceAgentID); team != nil { + if teamTask.TeamID != team.ID { + return nil, nil, fmt.Errorf("team_task_id does not belong to your team") + } + // Check team access for the end-user/channel + userID := store.UserIDFromContext(ctx) + channel := ToolChannelFromCtx(ctx) + if err := checkTeamAccess(team.Settings, userID, channel); err != nil { + return nil, nil, fmt.Errorf("team access denied: %w", err) + } + } + + // Auto-populate task description from spawn prompt if empty. + // This ensures the task board has full context for audit/visibility + // without relying on the LLM to set description at task creation time. + if teamTask.Description == "" && opts.Task != "" { + _ = dm.teamStore.UpdateTask(ctx, opts.TeamTaskID, map[string]any{ + "description": opts.Task, + }) + } + } + linkCount := dm.ActiveCountForLink(sourceAgentID, targetAgent.ID) if link.MaxConcurrent > 0 && linkCount >= link.MaxConcurrent { return nil, nil, fmt.Errorf("delegation link to %q is at capacity (%d/%d active). Try again later or handle the task yourself", diff --git a/internal/tools/delegate_events.go b/internal/tools/delegate_events.go index d0b315eb..ac4f8f4a 100644 --- a/internal/tools/delegate_events.go +++ b/internal/tools/delegate_events.go @@ -43,6 +43,15 @@ func formatDelegateAnnounce(task *DelegationTask, artifacts *DelegateArtifacts, // Render each delegation result for i, r := range artifacts.Results { msg += fmt.Sprintf("--- Result from %q ---\n%s\n", r.AgentKey, r.Content) + if len(r.Deliverables) > 0 { + for _, d := range r.Deliverables { + preview := d + if len(preview) > 4000 { + preview = preview[:4000] + "\n[...truncated, full content in team_tasks]" + } + msg += fmt.Sprintf("\n[Deliverable]\n%s\n", preview) + } + } if r.HasMedia { msg += "[media file(s) attached — will be delivered automatically. Do NOT recreate or call create_image.]\n" } diff --git a/internal/tools/delegate_policy.go b/internal/tools/delegate_policy.go index 4b22e7a4..71f701f7 100644 --- a/internal/tools/delegate_policy.go +++ b/internal/tools/delegate_policy.go @@ -33,6 +33,78 @@ func checkUserPermission(settings json.RawMessage, userID string) error { return nil } +// teamAccessSettings defines access control rules stored in agent_teams.settings JSONB. +// Empty/nil lists mean "no restriction". Deny lists take precedence over allow lists. +type teamAccessSettings struct { + AllowUserIDs []string `json:"allow_user_ids"` + DenyUserIDs []string `json:"deny_user_ids"` + AllowChannels []string `json:"allow_channels"` + DenyChannels []string `json:"deny_channels"` +} + +// checkTeamAccess validates whether a user/channel combination is authorized +// for team operations. Returns nil if access is allowed. +// System channels ("delegate", "system") always pass. +// Empty settings = open access (no restrictions). +func checkTeamAccess(settings json.RawMessage, userID, channel string) error { + if len(settings) == 0 || string(settings) == "{}" { + return nil + } + var s teamAccessSettings + if json.Unmarshal(settings, &s) != nil { + return nil // malformed = fail open + } + + // System/internal access always allowed + if channel == "delegate" || channel == "system" { + return nil + } + + // User check: deny > allow + if userID != "" { + for _, denied := range s.DenyUserIDs { + if denied == userID { + return fmt.Errorf("user not authorized for this team") + } + } + if len(s.AllowUserIDs) > 0 { + found := false + for _, allowed := range s.AllowUserIDs { + if allowed == userID { + found = true + break + } + } + if !found { + return fmt.Errorf("user not authorized for this team") + } + } + } + + // Channel check: deny > allow + if channel != "" { + for _, denied := range s.DenyChannels { + if denied == channel { + return fmt.Errorf("channel %q not authorized for this team", channel) + } + } + if len(s.AllowChannels) > 0 { + found := false + for _, allowed := range s.AllowChannels { + if allowed == channel { + found = true + break + } + } + if !found { + return fmt.Errorf("channel %q not authorized for this team", channel) + } + } + } + + return nil +} + func parseMaxDelegationLoad(otherConfig json.RawMessage) int { if len(otherConfig) == 0 { return defaultMaxDelegationLoad diff --git a/internal/tools/delegate_state.go b/internal/tools/delegate_state.go index b39649d0..38ccd1d9 100644 --- a/internal/tools/delegate_state.go +++ b/internal/tools/delegate_state.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "strings" "time" "github.com/google/uuid" @@ -127,12 +128,23 @@ func (dm *DelegateManager) flushCompletedSessions() { // Called after a delegation finishes successfully. Errors are logged but not fatal. // On success, flushes all tracked delegate sessions (task done = context no longer needed). // Also persists a team message record for audit trail / visualization. -func (dm *DelegateManager) autoCompleteTeamTask(task *DelegationTask, resultContent string) { +func (dm *DelegateManager) autoCompleteTeamTask(task *DelegationTask, resultContent string, deliverables []string) { if dm.teamStore == nil || task.TeamTaskID == uuid.Nil { return } - _ = dm.teamStore.ClaimTask(context.Background(), task.TeamTaskID, task.TargetAgentID) - if err := dm.teamStore.CompleteTask(context.Background(), task.TeamTaskID, resultContent); err != nil { + + // Use actual deliverables (tool outputs) as task result when available, + // falling back to the LLM's summary response. + taskResult := resultContent + if len(deliverables) > 0 { + taskResult = strings.Join(deliverables, "\n\n---\n\n") + if len(taskResult) > 100_000 { + taskResult = taskResult[:100_000] + "\n\n[truncated]" + } + } + + _ = dm.teamStore.ClaimTask(context.Background(), task.TeamTaskID, task.TargetAgentID, task.TeamID) + if err := dm.teamStore.CompleteTask(context.Background(), task.TeamTaskID, task.TeamID, taskResult); err != nil { slog.Warn("delegate: failed to auto-complete team task", "task_id", task.TeamTaskID, "delegation_id", task.ID, "error", err) } else { diff --git a/internal/tools/evaluate_loop_tool.go b/internal/tools/evaluate_loop_tool.go index fe244dc4..fb6482b8 100644 --- a/internal/tools/evaluate_loop_tool.go +++ b/internal/tools/evaluate_loop_tool.go @@ -145,8 +145,10 @@ func (t *EvaluateLoopTool) Execute(ctx context.Context, args map[string]interfac if isApproved(evalResult.Content) { // Auto-complete team task on the final successful round. if teamTaskID != uuid.Nil && t.manager.teamStore != nil { - _ = t.manager.teamStore.ClaimTask(ctx, teamTaskID, uuid.Nil) - _ = t.manager.teamStore.CompleteTask(ctx, teamTaskID, lastOutput) + if teamTask, getErr := t.manager.teamStore.GetTask(ctx, teamTaskID); getErr == nil { + _ = t.manager.teamStore.ClaimTask(ctx, teamTaskID, uuid.Nil, teamTask.TeamID) + _ = t.manager.teamStore.CompleteTask(ctx, teamTaskID, teamTask.TeamID, lastOutput) + } } return NewResult(fmt.Sprintf( diff --git a/internal/tools/filesystem.go b/internal/tools/filesystem.go index e2f93a2d..23de3849 100644 --- a/internal/tools/filesystem.go +++ b/internal/tools/filesystem.go @@ -277,12 +277,14 @@ func resolvePath(path, workspace string, restrict bool) (string, error) { } real = resolved } else { - // Truly non-existent file (not a symlink): resolve parent and re-validate. - parentReal, parentErr := filepath.EvalSymlinks(filepath.Dir(absResolved)) - if parentErr != nil { + // Truly non-existent file (not a symlink): walk up to find the + // deepest existing ancestor so nested new dirs (e.g. posts/file.md) + // are allowed as long as an ancestor is inside the workspace. + ancestorReal, ancestorErr := resolveThroughExistingAncestors(absResolved) + if ancestorErr != nil { return "", fmt.Errorf("access denied: cannot resolve path") } - real = filepath.Join(parentReal, filepath.Base(absResolved)) + real = ancestorReal } } else { // Permission error or other — reject. diff --git a/internal/tools/filesystem_write.go b/internal/tools/filesystem_write.go index 716bd802..6f76a499 100644 --- a/internal/tools/filesystem_write.go +++ b/internal/tools/filesystem_write.go @@ -124,6 +124,7 @@ func (t *WriteFileTool) Execute(ctx context.Context, args map[string]interface{} } result := SilentResult(fmt.Sprintf("File written: %s (%d bytes)", path, len(content))) + result.Deliverable = content if deliver { result.Media = []string{resolved} } @@ -141,6 +142,7 @@ func (t *WriteFileTool) executeInSandbox(ctx context.Context, path, content, san } result := SilentResult(fmt.Sprintf("File written: %s (%d bytes)", path, len(content))) + result.Deliverable = content if deliver { // Sandbox workspace is bind-mounted — resolve to host path for delivery workspace := ToolWorkspaceFromCtx(ctx) diff --git a/internal/tools/result.go b/internal/tools/result.go index 642eb837..8de2695d 100644 --- a/internal/tools/result.go +++ b/internal/tools/result.go @@ -14,6 +14,11 @@ type Result struct { // Media holds file paths to forward as output (e.g. images from delegation). Media []string `json:"-"` + // Deliverable holds the primary work output from this tool execution. + // Used to capture actual content (e.g. written file text, image prompt) for team + // task results instead of relying on the LLM's summary response. + Deliverable string `json:"-"` + // Usage holds token usage from tools that make internal LLM calls (e.g. read_image). // When set, the agent loop records these on the tool span for tracing. Usage *providers.Usage `json:"-"` diff --git a/internal/tools/team_message_tool.go b/internal/tools/team_message_tool.go index 3dff4071..62409137 100644 --- a/internal/tools/team_message_tool.go +++ b/internal/tools/team_message_tool.go @@ -82,6 +82,22 @@ func (t *TeamMessageTool) executeSend(ctx context.Context, args map[string]inter return ErrorResult(err.Error()) } + // Validate recipient is in the same team (prevent cross-team messaging). + members, err := t.manager.teamStore.ListMembers(ctx, team.ID) + if err != nil { + return ErrorResult("failed to verify team membership: " + err.Error()) + } + isMember := false + for _, m := range members { + if m.AgentID == toAgentID { + isMember = true + break + } + } + if !isMember { + return ErrorResult(fmt.Sprintf("agent %q is not a member of your team", toKey)) + } + // Persist to DB msg := &store.TeamMessageData{ TeamID: team.ID, diff --git a/internal/tools/team_tasks_tool.go b/internal/tools/team_tasks_tool.go index 615ea75c..180e7458 100644 --- a/internal/tools/team_tasks_tool.go +++ b/internal/tools/team_tasks_tool.go @@ -133,6 +133,11 @@ func (t *TeamTasksTool) executeList(ctx context.Context, args map[string]interfa } func (t *TeamTasksTool) executeGet(ctx context.Context, args map[string]interface{}) *Result { + team, _, err := t.manager.resolveTeam(ctx) + if err != nil { + return ErrorResult(err.Error()) + } + taskIDStr, _ := args["task_id"].(string) if taskIDStr == "" { return ErrorResult("task_id is required for get action") @@ -146,6 +151,9 @@ func (t *TeamTasksTool) executeGet(ctx context.Context, args map[string]interfac if err != nil { return ErrorResult("failed to get task: " + err.Error()) } + if task.TeamID != team.ID { + return ErrorResult("task does not belong to your team") + } // Truncate result for context protection (full result in DB) const maxResultRunes = 8000 @@ -253,7 +261,7 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]inter } func (t *TeamTasksTool) executeClaim(ctx context.Context, args map[string]interface{}) *Result { - _, agentID, err := t.manager.resolveTeam(ctx) + team, agentID, err := t.manager.resolveTeam(ctx) if err != nil { return ErrorResult(err.Error()) } @@ -267,7 +275,7 @@ func (t *TeamTasksTool) executeClaim(ctx context.Context, args map[string]interf return ErrorResult("invalid task_id") } - if err := t.manager.teamStore.ClaimTask(ctx, taskID, agentID); err != nil { + if err := t.manager.teamStore.ClaimTask(ctx, taskID, agentID, team.ID); err != nil { return ErrorResult("failed to claim task: " + err.Error()) } @@ -275,7 +283,7 @@ func (t *TeamTasksTool) executeClaim(ctx context.Context, args map[string]interf } func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]interface{}) *Result { - _, agentID, err := t.manager.resolveTeam(ctx) + team, agentID, err := t.manager.resolveTeam(ctx) if err != nil { return ErrorResult(err.Error()) } @@ -297,19 +305,16 @@ func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]int // Auto-claim if the task is still pending (saves an extra tool call). // ClaimTask is atomic — only one agent can succeed, others get an error. // Ignore claim error: task may already be in_progress (claimed by us or someone else). - _ = t.manager.teamStore.ClaimTask(ctx, taskID, agentID) + _ = t.manager.teamStore.ClaimTask(ctx, taskID, agentID, team.ID) - if err := t.manager.teamStore.CompleteTask(ctx, taskID, result); err != nil { + if err := t.manager.teamStore.CompleteTask(ctx, taskID, team.ID, result); err != nil { return ErrorResult("failed to complete task: " + err.Error()) } - // Resolve team for event payload - if team, _, teamErr := t.manager.resolveTeam(ctx); teamErr == nil { - t.manager.broadcastTeamEvent(protocol.EventTeamTaskCompleted, map[string]string{ - "team_id": team.ID.String(), - "task_id": taskIDStr, - }) - } + t.manager.broadcastTeamEvent(protocol.EventTeamTaskCompleted, map[string]string{ + "team_id": team.ID.String(), + "task_id": taskIDStr, + }) return NewResult(fmt.Sprintf("Task %s completed. Dependent tasks have been unblocked.", taskIDStr)) } diff --git a/internal/tools/team_tool_manager.go b/internal/tools/team_tool_manager.go index 6a0ab15b..0fd30c6e 100644 --- a/internal/tools/team_tool_manager.go +++ b/internal/tools/team_tool_manager.go @@ -3,6 +3,8 @@ package tools import ( "context" "fmt" + "sync" + "time" "github.com/google/uuid" @@ -10,13 +12,23 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/store" ) +const teamCacheTTL = 5 * time.Minute + +// teamCacheEntry wraps cached team data with a timestamp for TTL expiration. +type teamCacheEntry struct { + team *store.TeamData + cachedAt time.Time +} + // TeamToolManager is the shared backend for team_tasks and team_message tools. // It resolves the calling agent's team from context and provides access to // the team store, agent store, and message bus. +// Includes a TTL cache for team data to avoid DB queries on every tool call. type TeamToolManager struct { teamStore store.TeamStore agentStore store.AgentStore msgBus *bus.MessageBus + teamCache sync.Map // agentID (uuid.UUID) → *teamCacheEntry } func NewTeamToolManager(teamStore store.TeamStore, agentStore store.AgentStore, msgBus *bus.MessageBus) *TeamToolManager { @@ -24,12 +36,30 @@ func NewTeamToolManager(teamStore store.TeamStore, agentStore store.AgentStore, } // resolveTeam returns the team that the calling agent belongs to. +// Uses a TTL cache to avoid repeated DB queries. Access control +// (user/channel) is checked on every call regardless of cache hit. func (m *TeamToolManager) resolveTeam(ctx context.Context) (*store.TeamData, uuid.UUID, error) { agentID := store.AgentIDFromContext(ctx) if agentID == uuid.Nil { return nil, uuid.Nil, fmt.Errorf("no agent context — team tools require managed mode") } + // Check cache first + if entry, ok := m.teamCache.Load(agentID); ok { + ce := entry.(*teamCacheEntry) + if time.Since(ce.cachedAt) < teamCacheTTL { + // Cache hit — still check access (user/channel vary per call) + userID := store.UserIDFromContext(ctx) + channel := ToolChannelFromCtx(ctx) + if err := checkTeamAccess(ce.team.Settings, userID, channel); err != nil { + return nil, uuid.Nil, err + } + return ce.team, agentID, nil + } + m.teamCache.Delete(agentID) // expired + } + + // Cache miss → DB team, err := m.teamStore.GetTeamForAgent(ctx, agentID) if err != nil { return nil, uuid.Nil, fmt.Errorf("failed to resolve team: %w", err) @@ -38,9 +68,26 @@ func (m *TeamToolManager) resolveTeam(ctx context.Context) (*store.TeamData, uui return nil, uuid.Nil, fmt.Errorf("this agent is not part of any team") } + // Store in cache + m.teamCache.Store(agentID, &teamCacheEntry{team: team, cachedAt: time.Now()}) + + // Check access + userID := store.UserIDFromContext(ctx) + channel := ToolChannelFromCtx(ctx) + if err := checkTeamAccess(team.Settings, userID, channel); err != nil { + return nil, uuid.Nil, err + } + return team, agentID, nil } +// InvalidateTeam clears all cached team data. +// Called when team membership, settings, or links change. +// Full clear is acceptable because team mutations are rare (admin-initiated). +func (m *TeamToolManager) InvalidateTeam() { + m.teamCache = sync.Map{} +} + // resolveAgentByKey looks up an agent by key and returns its UUID. func (m *TeamToolManager) resolveAgentByKey(key string) (uuid.UUID, error) { ag, err := m.agentStore.GetByKey(context.Background(), key) diff --git a/internal/tools/tts.go b/internal/tools/tts.go index adb350f2..0ed088b2 100644 --- a/internal/tools/tts.go +++ b/internal/tools/tts.go @@ -100,6 +100,8 @@ func (t *TtsTool) Execute(ctx context.Context, args map[string]interface{}) *Res voiceTag = "[[audio_as_voice]]\n" } - content := fmt.Sprintf("%sMEDIA:%s", voiceTag, audioPath) - return &Result{ForLLM: content} + forLLM := fmt.Sprintf("%sMEDIA:%s", voiceTag, audioPath) + r := &Result{ForLLM: forLLM} + r.Deliverable = fmt.Sprintf("[Generated audio: %s]\nText: %s", filepath.Base(audioPath), text) + return r } diff --git a/pkg/protocol/methods.go b/pkg/protocol/methods.go index ce03c85d..e4777ffe 100644 --- a/pkg/protocol/methods.go +++ b/pkg/protocol/methods.go @@ -106,6 +106,8 @@ const ( MethodTeamsTaskList = "teams.tasks.list" MethodTeamsMembersAdd = "teams.members.add" MethodTeamsMembersRemove = "teams.members.remove" + MethodTeamsUpdate = "teams.update" + MethodTeamsKnownUsers = "teams.known_users" ) // Delegation history (managed mode) diff --git a/ui/web/src/api/protocol.ts b/ui/web/src/api/protocol.ts index 5e4a8305..fb3f5699 100644 --- a/ui/web/src/api/protocol.ts +++ b/ui/web/src/api/protocol.ts @@ -131,6 +131,8 @@ export const Methods = { TEAMS_TASK_LIST: "teams.tasks.list", TEAMS_MEMBERS_ADD: "teams.members.add", TEAMS_MEMBERS_REMOVE: "teams.members.remove", + TEAMS_UPDATE: "teams.update", + TEAMS_KNOWN_USERS: "teams.known_users", // Delegation history (managed mode) DELEGATIONS_LIST: "delegations.list", diff --git a/ui/web/src/constants/channels.ts b/ui/web/src/constants/channels.ts new file mode 100644 index 00000000..510cfb7e --- /dev/null +++ b/ui/web/src/constants/channels.ts @@ -0,0 +1,7 @@ +export const CHANNEL_TYPES = [ + { value: "telegram", label: "Telegram" }, + { value: "discord", label: "Discord" }, + { value: "feishu", label: "Feishu / Lark" }, + { value: "zalo_oa", label: "Zalo OA" }, + { value: "whatsapp", label: "WhatsApp" }, +] as const; diff --git a/ui/web/src/pages/channels/channel-instance-form-dialog.tsx b/ui/web/src/pages/channels/channel-instance-form-dialog.tsx index 8ad870f4..20b340c6 100644 --- a/ui/web/src/pages/channels/channel-instance-form-dialog.tsx +++ b/ui/web/src/pages/channels/channel-instance-form-dialog.tsx @@ -22,14 +22,7 @@ import type { AgentData } from "@/types/agent"; import { slugify, isValidSlug } from "@/lib/slug"; import { credentialsSchema, configSchema } from "./channel-schemas"; import { ChannelFields } from "./channel-fields"; - -const CHANNEL_TYPES = [ - { value: "telegram", label: "Telegram" }, - { value: "discord", label: "Discord" }, - { value: "feishu", label: "Feishu / Lark" }, - { value: "zalo_oa", label: "Zalo OA" }, - { value: "whatsapp", label: "WhatsApp" }, -] as const; +import { CHANNEL_TYPES } from "@/constants/channels"; interface ChannelInstanceFormDialogProps { open: boolean; diff --git a/ui/web/src/pages/teams/hooks/use-teams.ts b/ui/web/src/pages/teams/hooks/use-teams.ts index 1a92d4f7..b4003daf 100644 --- a/ui/web/src/pages/teams/hooks/use-teams.ts +++ b/ui/web/src/pages/teams/hooks/use-teams.ts @@ -1,7 +1,7 @@ import { useState, useCallback } from "react"; import { useWs } from "@/hooks/use-ws"; import { Methods } from "@/api/protocol"; -import type { TeamData, TeamMemberData, TeamTaskData } from "@/types/team"; +import type { TeamData, TeamMemberData, TeamTaskData, TeamAccessSettings } from "@/types/team"; export function useTeams() { const ws = useWs(); @@ -80,5 +80,23 @@ export function useTeams() { [ws], ); - return { teams, loading, load, createTeam, deleteTeam, getTeam, getTeamTasks, addMember, removeMember }; + const updateTeamSettings = useCallback( + async (teamId: string, settings: TeamAccessSettings) => { + await ws.call(Methods.TEAMS_UPDATE, { teamId, settings }); + }, + [ws], + ); + + const getKnownUsers = useCallback( + async (teamId: string): Promise => { + const res = await ws.call<{ users: string[] }>( + Methods.TEAMS_KNOWN_USERS, + { teamId }, + ); + return res.users ?? []; + }, + [ws], + ); + + return { teams, loading, load, createTeam, deleteTeam, getTeam, getTeamTasks, addMember, removeMember, updateTeamSettings, getKnownUsers }; } diff --git a/ui/web/src/pages/teams/team-detail-page.tsx b/ui/web/src/pages/teams/team-detail-page.tsx index d99c4ffd..e7d80363 100644 --- a/ui/web/src/pages/teams/team-detail-page.tsx +++ b/ui/web/src/pages/teams/team-detail-page.tsx @@ -8,6 +8,7 @@ import { useTeams } from "./hooks/use-teams"; import { TeamMembersTab } from "./team-members-tab"; import { TeamTasksTab } from "./team-tasks-tab"; import { TeamDelegationsTab } from "./team-delegations-tab"; +import { TeamSettingsTab } from "./team-settings-tab"; import type { TeamData, TeamMemberData } from "@/types/team"; interface TeamDetailPageProps { @@ -110,6 +111,7 @@ export function TeamDetailPage({ teamId, onBack }: TeamDetailPageProps) { Members Tasks Delegations + Settings @@ -128,6 +130,10 @@ export function TeamDetailPage({ teamId, onBack }: TeamDetailPageProps) { + + + + diff --git a/ui/web/src/pages/teams/team-settings-tab.tsx b/ui/web/src/pages/teams/team-settings-tab.tsx new file mode 100644 index 00000000..ba40ca55 --- /dev/null +++ b/ui/web/src/pages/teams/team-settings-tab.tsx @@ -0,0 +1,197 @@ +import { useState, useEffect, useCallback } from "react"; +import { Button } from "@/components/ui/button"; +import { Badge } from "@/components/ui/badge"; +import { Combobox } from "@/components/ui/combobox"; +import { X, Save, Check } from "lucide-react"; +import { CHANNEL_TYPES } from "@/constants/channels"; +import type { TeamData, TeamAccessSettings } from "@/types/team"; +import { useTeams } from "./hooks/use-teams"; + +interface TeamSettingsTabProps { + teamId: string; + team: TeamData; + onSaved: () => void; +} + +function MultiSelect({ + options, + selected, + onChange, + placeholder, +}: { + options: { value: string; label?: string }[]; + selected: string[]; + onChange: (values: string[]) => void; + placeholder: string; +}) { + return ( +
+ { + if (val && !selected.includes(val)) { + onChange([...selected, val]); + } + }} + options={options.filter((o) => !selected.includes(o.value))} + placeholder={placeholder} + /> + {selected.length > 0 && ( +
+ {selected.map((id) => ( + + {options.find((o) => o.value === id)?.label ?? id} + + + ))} +
+ )} +
+ ); +} + +export function TeamSettingsTab({ teamId, team, onSaved }: TeamSettingsTabProps) { + const { updateTeamSettings, getKnownUsers } = useTeams(); + const [knownUsers, setKnownUsers] = useState([]); + + // Parse initial settings + const initial = (team.settings ?? {}) as TeamAccessSettings; + const [allowUserIds, setAllowUserIds] = useState(initial.allow_user_ids ?? []); + const [denyUserIds, setDenyUserIds] = useState(initial.deny_user_ids ?? []); + const [allowChannels, setAllowChannels] = useState(initial.allow_channels ?? []); + const [denyChannels, setDenyChannels] = useState(initial.deny_channels ?? []); + + const [saving, setSaving] = useState(false); + const [saved, setSaved] = useState(false); + const [error, setError] = useState(null); + + // Load known users for combobox + useEffect(() => { + getKnownUsers(teamId).then(setKnownUsers).catch(() => {}); + }, [teamId, getKnownUsers]); + + // Reset when team changes + useEffect(() => { + const s = (team.settings ?? {}) as TeamAccessSettings; + setAllowUserIds(s.allow_user_ids ?? []); + setDenyUserIds(s.deny_user_ids ?? []); + setAllowChannels(s.allow_channels ?? []); + setDenyChannels(s.deny_channels ?? []); + setSaved(false); + setError(null); + }, [team]); + + const handleSave = useCallback(async () => { + setSaving(true); + setError(null); + setSaved(false); + try { + const settings: TeamAccessSettings = {}; + if (allowUserIds.length > 0) settings.allow_user_ids = allowUserIds; + if (denyUserIds.length > 0) settings.deny_user_ids = denyUserIds; + if (allowChannels.length > 0) settings.allow_channels = allowChannels; + if (denyChannels.length > 0) settings.deny_channels = denyChannels; + await updateTeamSettings(teamId, settings); + setSaved(true); + onSaved(); + setTimeout(() => setSaved(false), 3000); + } catch (err) { + setError(err instanceof Error ? err.message : "Failed to save"); + } finally { + setSaving(false); + } + }, [teamId, allowUserIds, denyUserIds, allowChannels, denyChannels, updateTeamSettings, onSaved]); + + const userOptions = knownUsers.map((u) => ({ value: u, label: u })); + const channelOptions = CHANNEL_TYPES.map((c) => ({ value: c.value, label: c.label })); + + return ( +
+ {/* User Access Control */} +
+

User Access Control

+
+
+ +

+ If set, only these users can trigger team workflows. Empty = all allowed. +

+ +
+
+ +

+ Always blocked, overrides allow list. +

+ +
+
+
+ + {/* Channel Restrictions */} +
+

Channel Restrictions

+
+
+ +

+ If set, only these channels can activate team workflows. Empty = all allowed. +

+ +
+
+ +

+ Messages from these channels are always blocked. +

+ +
+
+
+ + {/* Save button */} +
+ + {error && {error}} +
+
+ ); +} diff --git a/ui/web/src/types/team.ts b/ui/web/src/types/team.ts index 4ffa53ac..08ac0dfb 100644 --- a/ui/web/src/types/team.ts +++ b/ui/web/src/types/team.ts @@ -1,5 +1,12 @@ /** Team data types matching Go internal/store/team_store.go */ +export interface TeamAccessSettings { + allow_user_ids?: string[]; + deny_user_ids?: string[]; + allow_channels?: string[]; + deny_channels?: string[]; +} + export interface TeamData { id: string; name: string;