diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index cca6b2b6..8f37f87c 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -336,6 +336,10 @@ func wireManagedExtras( Stream: req.Stream, ExtraSystemPrompt: req.ExtraSystemPrompt, MaxIterations: req.MaxIterations, + DelegationID: req.DelegationID, + TeamID: req.TeamID, + TeamTaskID: req.TeamTaskID, + ParentAgentID: req.ParentAgentID, }) if err != nil { return nil, err diff --git a/docs/13-ws-team-events.md b/docs/13-ws-team-events.md new file mode 100644 index 00000000..c8324dc2 --- /dev/null +++ b/docs/13-ws-team-events.md @@ -0,0 +1,501 @@ +# WebSocket Team & Delegation Events + +Complete reference for all WS events related to team agent operations, delegation lifecycle, and admin CRUD. + +All events are delivered as JSON frames via the WebSocket protocol: +```json +{"type": "event", "event": "", "payload": { ... }} +``` + +Events are emitted via `msgBus.Broadcast(bus.Event{})` and forwarded to all connected WS clients (filtered by gateway subscriber at `server.go`). + +--- + +## Event Catalog + +### Delegation Lifecycle Events + +#### `delegation.started` +Emitted when a lead agent initiates a delegation to a member agent. + +```json +{ + "delegation_id": "a1b2c3d4", + "source_agent_id": "019c839b-...", + "source_agent_key": "default", + "source_display_name": "Default Agent", + "target_agent_id": "019ca748-...", + "target_agent_key": "tieu-la", + "target_display_name": "Tieu La", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456", + "mode": "async", + "task": "Create Instagram image for new product", + "team_id": "019c9503-...", + "team_task_id": "019ca84f-...", + "status": "running", + "created_at": "2026-03-05T10:00:00Z" +} +``` + +#### `delegation.completed` +Emitted when a delegation finishes successfully (quality gates passed). + +Same payload as `delegation.started` with: +- `status`: `"completed"` +- `elapsed_ms`: total duration in milliseconds + +#### `delegation.failed` +Emitted when a delegation fails (agent error or quality gate rejection). + +Same payload as `delegation.started` with: +- `status`: `"failed"` +- `error`: error message string +- `elapsed_ms`: total duration + +#### `delegation.cancelled` +Emitted when a delegation is cancelled (via `/stopall`, team task cancel, or direct cancel). + +Same payload as `delegation.started` with: +- `status`: `"cancelled"` +- `elapsed_ms`: total duration + +#### `delegation.progress` +Emitted periodically (~30s) for active async delegations. Groups all active delegations from the same source agent. + +```json +{ + "source_agent_id": "019c839b-...", + "source_agent_key": "default", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456", + "team_id": "019c9503-...", + "active_delegations": [ + { + "delegation_id": "a1b2c3d4", + "target_agent_key": "tieu-la", + "target_display_name": "Tieu La", + "elapsed_ms": 45000, + "team_task_id": "019ca84f-..." + }, + { + "delegation_id": "e5f6g7h8", + "target_agent_key": "tieu-ngon", + "target_display_name": "Tieu Ngon", + "elapsed_ms": 30000, + "team_task_id": "019ca850-..." + } + ] +} +``` + +#### `delegation.accumulated` +Emitted when an async delegation completes but sibling delegations are still running. The result is accumulated and will be announced when all siblings finish. + +```json +{ + "delegation_id": "a1b2c3d4", + "source_agent_id": "019c839b-...", + "source_agent_key": "default", + "target_agent_key": "tieu-la", + "target_display_name": "Tieu La", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456", + "team_id": "019c9503-...", + "team_task_id": "019ca84f-...", + "siblings_remaining": 1, + "elapsed_ms": 45300 +} +``` + +#### `delegation.announce` +Emitted when the last sibling delegation completes and all accumulated results are sent back to the lead agent. + +```json +{ + "source_agent_id": "019c839b-...", + "source_agent_key": "default", + "source_display_name": "Default Agent", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456", + "team_id": "019c9503-...", + "results": [ + { + "agent_key": "tieu-la", + "display_name": "Tieu La", + "has_media": true, + "content_preview": "Created Instagram post image..." + }, + { + "agent_key": "tieu-ngon", + "display_name": "Tieu Ngon", + "has_media": false, + "content_preview": "Wrote caption for you..." + } + ], + "completed_task_ids": ["019ca84f-...", "019ca850-..."], + "total_elapsed_ms": 52000, + "has_media": true +} +``` + +#### `delegation.quality_gate.retry` +Emitted when a quality gate rejects a delegation result and triggers a retry. + +```json +{ + "delegation_id": "a1b2c3d4", + "target_agent_key": "tieu-la", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456", + "team_id": "019c9503-...", + "team_task_id": "019ca84f-...", + "gate_type": "agent", + "attempt": 2, + "max_retries": 3, + "feedback": "Image aspect ratio should be 4:5..." +} +``` + +--- + +### Team Task Events + +#### `team.task.created` +Emitted when a new team task is created (manual or auto-created by delegation). + +```json +{ + "team_id": "019c9503-...", + "task_id": "019ca84f-...", + "subject": "Create Instagram image", + "status": "pending", + "owner_agent_key": "", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456", + "timestamp": "2026-03-05T10:00:00Z" +} +``` + +#### `team.task.claimed` +Emitted when an agent claims a task (manual or auto-claim before delegation). + +```json +{ + "team_id": "019c9503-...", + "task_id": "019ca84f-...", + "status": "in_progress", + "owner_agent_key": "tieu-la", + "owner_display_name": "Tieu La", + "user_id": "user123", + "channel": "delegate", + "chat_id": "-100123456", + "timestamp": "2026-03-05T10:00:01Z" +} +``` + +#### `team.task.completed` +Emitted when a task is completed (manual or auto-completed by delegation). + +```json +{ + "team_id": "019c9503-...", + "task_id": "019ca84f-...", + "status": "completed", + "owner_agent_key": "tieu-la", + "owner_display_name": "Tieu La", + "user_id": "user123", + "channel": "delegate", + "chat_id": "-100123456", + "timestamp": "2026-03-05T10:00:45Z" +} +``` + +#### `team.task.cancelled` +Emitted when a task is cancelled. Separated from `team.task.completed` for correct semantics. + +```json +{ + "team_id": "019c9503-...", + "task_id": "019ca84f-...", + "status": "cancelled", + "reason": "Task no longer needed", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456", + "timestamp": "2026-03-05T10:01:00Z" +} +``` + +--- + +### Team Message Events + +#### `team.message.sent` +Emitted when an agent sends a message to another agent or broadcasts to the team. + +```json +{ + "team_id": "019c9503-...", + "from_agent_key": "default", + "from_display_name": "Default Agent", + "to_agent_key": "tieu-la", + "to_display_name": "Tieu La", + "message_type": "chat", + "preview": "Please create an Instagram image...", + "task_id": "", + "user_id": "user123", + "channel": "telegram", + "chat_id": "-100123456" +} +``` + +For broadcast messages: `to_agent_key = "broadcast"`, `to_display_name = ""`. + +--- + +### Team CRUD Events (Admin) + +These events are emitted from RPC handlers when teams are managed via the Web UI. No routing context (user_id/channel/chat_id) since these are admin operations. + +#### `team.created` +```json +{ + "team_id": "019c9503-...", + "team_name": "Content Team", + "lead_agent_key": "default", + "lead_display_name": "Default Agent", + "member_count": 3 +} +``` + +#### `team.updated` +```json +{ + "team_id": "019c9503-...", + "team_name": "Content Team", + "changes": ["settings"] +} +``` + +#### `team.deleted` +```json +{ + "team_id": "019c9503-...", + "team_name": "Content Team" +} +``` + +#### `team.member.added` +```json +{ + "team_id": "019c9503-...", + "team_name": "Content Team", + "agent_id": "019ca748-...", + "agent_key": "tieu-la", + "display_name": "Tieu La", + "role": "member" +} +``` + +#### `team.member.removed` +```json +{ + "team_id": "019c9503-...", + "team_name": "Content Team", + "agent_id": "019ca748-...", + "agent_key": "tieu-la", + "display_name": "Tieu La" +} +``` + +--- + +### Agent Link Events (Admin) + +Emitted from RPC handlers when agent links are managed via the Web UI. + +#### `agent_link.created` +```json +{ + "link_id": "019cab12-...", + "source_agent_id": "019c839b-...", + "source_agent_key": "default", + "target_agent_id": "019ca748-...", + "target_agent_key": "tieu-la", + "direction": "outbound", + "team_id": "019c9503-...", + "status": "active" +} +``` + +#### `agent_link.updated` +```json +{ + "link_id": "019cab12-...", + "source_agent_key": "default", + "target_agent_key": "tieu-la", + "direction": "bidirectional", + "status": "active", + "changes": ["direction", "settings"] +} +``` + +#### `agent_link.deleted` +```json +{ + "link_id": "019cab12-...", + "source_agent_key": "default", + "target_agent_key": "tieu-la" +} +``` + +--- + +### Agent Events (Delegation Context) + +`AgentEvent` payloads (broadcast as `"event": "agent"`) now include optional delegation and routing context: + +**`tool.call` example** (member agent inside delegation): +```json +{ + "type": "tool.call", + "agentId": "tieu-la", + "runId": "delegate-a1b2c3d4", + "delegationId": "a1b2c3d4", + "teamId": "019c9503-...", + "teamTaskId": "019ca84f-...", + "parentAgentId": "default", + "userId": "user123", + "channel": "telegram", + "chatId": "-100123456", + "payload": {"name": "create_image", "id": "call_xxx"} +} +``` + +**`tool.result` example:** +```json +{ + "type": "tool.result", + "agentId": "tieu-la", + "runId": "delegate-a1b2c3d4", + "delegationId": "a1b2c3d4", + "teamId": "019c9503-...", + "teamTaskId": "019ca84f-...", + "parentAgentId": "default", + "userId": "user123", + "channel": "telegram", + "chatId": "-100123456", + "payload": {"name": "create_image", "id": "call_xxx", "is_error": false} +} +``` + +> **Note:** Tool arguments and result content are intentionally omitted from payloads to avoid leaking sensitive data over WS. Only tool name, call ID, and error status are included. + +**Agent event subtypes** (the `type` field inside the payload): + +| Constant | Type | Description | +|----------|------|-------------| +| `AgentEventRunStarted` | `run.started` | Agent run begins | +| `AgentEventRunCompleted` | `run.completed` | Agent run finished successfully | +| `AgentEventRunFailed` | `run.failed` | Agent run failed | +| `AgentEventRunRetrying` | `run.retrying` | Agent run retrying after error | +| `AgentEventToolCall` | `tool.call` | Agent calling a tool | +| `AgentEventToolResult` | `tool.result` | Tool execution completed | +| *(chat events)* | `chunk` | Streaming text chunk | +| *(chat events)* | `thinking` | Extended thinking content | + +**Note:** When `Stream: true`, `chunk` and `thinking` are emitted incrementally (one event per streamed fragment). When `Stream: false` (e.g. delegate runs), they are emitted as a single event containing the full content after the LLM response is received. Both paths carry full delegation context. + +Fields present only when the agent is running inside a delegation: +- `delegationId` — correlation ID for this delegation +- `teamId` — team scope (if team-based) +- `teamTaskId` — associated team task +- `parentAgentId` — lead agent key that initiated the delegation + +Fields always present when available: +- `userId` — scoped user ID (group chats: `"group:{channel}:{chatID}"`) +- `channel` — origin channel (telegram, discord, web, etc.) +- `chatId` — origin chat/conversation ID + +Client can distinguish lead vs member agent events: +- `parentAgentId` absent → lead agent event +- `parentAgentId` present → member agent event (delegation) + +--- + +## Event Flow Timeline + +``` +User sends "Create Instagram post" to Default Agent (lead) + | + v +[agent] run.started agentId=default +[agent] chunk agentId=default, content="Let me assign..." +[agent] tool.call agentId=default, tool=delegate + | + |-- [delegation.started] target=tieu-la, task="Create image", mode=async + |-- [delegation.started] target=tieu-ngon, task="Write caption", mode=async + | + | (member agents run in parallel) + | + |-- [agent] run.started agentId=tieu-la, delegationId=xxx, parentAgentId=default + |-- [agent] run.started agentId=tieu-ngon, delegationId=yyy, parentAgentId=default + | + |-- [agent] tool.call agentId=tieu-la, tool=create_image, delegationId=xxx + |-- [agent] tool.result agentId=tieu-la, delegationId=xxx + | + |-- [delegation.progress] active=[{tieu-la, 30s}, {tieu-ngon, 30s}] + | + |-- [agent] run.completed agentId=tieu-la, delegationId=xxx + |-- [delegation.completed] target=tieu-la, elapsed_ms=35000 + |-- [delegation.accumulated] target=tieu-la, siblings_remaining=1 + | + |-- [agent] run.completed agentId=tieu-ngon, delegationId=yyy + |-- [delegation.completed] target=tieu-ngon, elapsed_ms=42000 + |-- [delegation.announce] results=[{tieu-la, has_media}, {tieu-ngon}] + | + | (lead receives announce, processes results) + | + |-- [agent] run.started agentId=default + |-- [agent] chunk agentId=default, content="Here are the results..." + +-- [agent] run.completed agentId=default +``` + +--- + +## Constants Reference + +All event name constants are defined in `pkg/protocol/events.go`: + +| Constant | Event Name | +|----------|-----------| +| `EventDelegationStarted` | `delegation.started` | +| `EventDelegationCompleted` | `delegation.completed` | +| `EventDelegationFailed` | `delegation.failed` | +| `EventDelegationCancelled` | `delegation.cancelled` | +| `EventDelegationProgress` | `delegation.progress` | +| `EventDelegationAccumulated` | `delegation.accumulated` | +| `EventDelegationAnnounce` | `delegation.announce` | +| `EventQualityGateRetry` | `delegation.quality_gate.retry` | +| `EventTeamTaskCreated` | `team.task.created` | +| `EventTeamTaskClaimed` | `team.task.claimed` | +| `EventTeamTaskCompleted` | `team.task.completed` | +| `EventTeamTaskCancelled` | `team.task.cancelled` | +| `EventTeamMessageSent` | `team.message.sent` | +| `EventTeamCreated` | `team.created` | +| `EventTeamUpdated` | `team.updated` | +| `EventTeamDeleted` | `team.deleted` | +| `EventTeamMemberAdded` | `team.member.added` | +| `EventTeamMemberRemoved` | `team.member.removed` | +| `EventAgentLinkCreated` | `agent_link.created` | +| `EventAgentLinkUpdated` | `agent_link.updated` | +| `EventAgentLinkDeleted` | `agent_link.deleted` | + +Typed payload structs are defined in `pkg/protocol/team_events.go`. diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 38b1a6d2..ecf33df7 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -28,7 +28,19 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { l.activeRuns.Add(1) defer l.activeRuns.Add(-1) - l.emit(AgentEvent{Type: protocol.AgentEventRunStarted, AgentID: l.id, RunID: req.RunID}) + // Per-run emit wrapper: enriches every AgentEvent with delegation + routing context. + emitRun := func(event AgentEvent) { + event.DelegationID = req.DelegationID + event.TeamID = req.TeamID + event.TeamTaskID = req.TeamTaskID + event.ParentAgentID = req.ParentAgentID + event.UserID = req.UserID + event.Channel = req.Channel + event.ChatID = req.ChatID + l.emit(event) + } + + emitRun(AgentEvent{Type: protocol.AgentEventRunStarted, AgentID: l.id, RunID: req.RunID}) // Create trace (managed mode only) var traceID uuid.UUID @@ -98,7 +110,7 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { } if err != nil { - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.AgentEventRunFailed, AgentID: l.id, RunID: req.RunID, @@ -119,7 +131,7 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { return nil, err } - l.emit(AgentEvent{Type: protocol.AgentEventRunCompleted, AgentID: l.id, RunID: req.RunID}) + emitRun(AgentEvent{Type: protocol.AgentEventRunCompleted, AgentID: l.id, RunID: req.RunID}) if !isChildTrace && l.traceCollector != nil && traceID != uuid.Nil { l.traceCollector.FinishTrace(ctx, traceID, store.TraceStatusCompleted, "", truncateStr(result.Content, 500)) } @@ -127,6 +139,18 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { } func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) { + // Per-run emit wrapper: enriches every AgentEvent with delegation + routing context. + emitRun := func(event AgentEvent) { + event.DelegationID = req.DelegationID + event.TeamID = req.TeamID + event.TeamTaskID = req.TeamTaskID + event.ParentAgentID = req.ParentAgentID + event.UserID = req.UserID + event.Channel = req.Channel + event.ChatID = req.ChatID + l.emit(event) + } + // Inject agent UUID into context for tool routing (managed mode) if l.agentUUID != uuid.Nil { ctx = store.WithAgentID(ctx, l.agentUUID) @@ -346,7 +370,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // Inject retry hook so channels can update placeholder on LLM retries. ctx = providers.WithRetryHook(ctx, func(attempt, maxAttempts int, err error) { - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.AgentEventRunRetrying, AgentID: l.id, RunID: req.RunID, @@ -408,7 +432,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) if req.Stream { resp, err = l.provider.ChatStream(ctx, chatReq, func(chunk providers.StreamChunk) { if chunk.Thinking != "" { - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.ChatEventThinking, AgentID: l.id, RunID: req.RunID, @@ -416,7 +440,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) }) } if chunk.Content != "" { - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.ChatEventChunk, AgentID: l.id, RunID: req.RunID, @@ -435,6 +459,26 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) l.emitLLMSpan(ctx, llmSpanStart, iteration, messages, resp, nil) + // For non-streaming responses, emit thinking and content as single events + if !req.Stream { + if resp.Thinking != "" { + emitRun(AgentEvent{ + Type: protocol.ChatEventThinking, + AgentID: l.id, + RunID: req.RunID, + Payload: map[string]string{"content": resp.Thinking}, + }) + } + if resp.Content != "" { + emitRun(AgentEvent{ + Type: protocol.ChatEventChunk, + AgentID: l.id, + RunID: req.RunID, + Payload: map[string]string{"content": resp.Content}, + }) + } + } + if resp.Usage != nil { totalUsage.PromptTokens += resp.Usage.PromptTokens totalUsage.CompletionTokens += resp.Usage.CompletionTokens @@ -547,7 +591,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) if len(resp.ToolCalls) == 1 { // Single tool: sequential — no goroutine overhead tc := resp.ToolCalls[0] - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.AgentEventToolCall, AgentID: l.id, RunID: req.RunID, @@ -592,7 +636,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } } - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.AgentEventToolResult, AgentID: l.id, RunID: req.RunID, @@ -647,7 +691,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // 1. Emit all tool.call events upfront (client sees all calls starting) for _, tc := range resp.ToolCalls { - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.AgentEventToolCall, AgentID: l.id, RunID: req.RunID, @@ -719,7 +763,7 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } } - l.emit(AgentEvent{ + emitRun(AgentEvent{ Type: protocol.AgentEventToolResult, AgentID: l.id, RunID: req.RunID, diff --git a/internal/agent/loop_types.go b/internal/agent/loop_types.go index cbb56b38..06c90f10 100644 --- a/internal/agent/loop_types.go +++ b/internal/agent/loop_types.go @@ -110,6 +110,17 @@ type AgentEvent struct { AgentID string `json:"agentId"` RunID string `json:"runId"` Payload interface{} `json:"payload,omitempty"` + + // Delegation context (omitempty — only present when agent runs inside a delegation) + DelegationID string `json:"delegationId,omitempty"` + TeamID string `json:"teamId,omitempty"` + TeamTaskID string `json:"teamTaskId,omitempty"` + ParentAgentID string `json:"parentAgentId,omitempty"` + + // Routing context (helps WS clients filter by user/channel) + UserID string `json:"userId,omitempty"` + Channel string `json:"channel,omitempty"` + ChatID string `json:"chatId,omitempty"` } // LoopConfig configures a new Loop. @@ -262,6 +273,12 @@ type RunRequest struct { TraceName string // override trace name (default: "chat ") TraceTags []string // additional tags for the trace (e.g. "cron") MaxIterations int // per-request override (0 = use agent default, must be lower) + + // Delegation context (set when running as a delegate agent) + DelegationID string // delegation ID for event correlation + TeamID string // team ID (if delegation is team-scoped) + TeamTaskID string // team task ID (if delegation has an associated task) + ParentAgentID string // parent agent key that initiated the delegation } // RunResult is the output of a completed agent run. diff --git a/internal/gateway/methods/agent_links.go b/internal/gateway/methods/agent_links.go index 92e819f7..cb49ff19 100644 --- a/internal/gateway/methods/agent_links.go +++ b/internal/gateway/methods/agent_links.go @@ -184,6 +184,26 @@ func (m *AgentLinksMethods) handleCreate(_ context.Context, client *gateway.Clie client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ "link": link, })) + + // Emit agent_link.created event + if m.msgBus != nil { + payload := protocol.AgentLinkCreatedPayload{ + LinkID: link.ID.String(), + SourceAgentID: sourceAgent.ID.String(), + SourceAgentKey: sourceAgent.AgentKey, + TargetAgentID: targetAgent.ID.String(), + TargetAgentKey: targetAgent.AgentKey, + Direction: direction, + Status: store.LinkStatusActive, + } + if link.TeamID != nil { + payload.TeamID = link.TeamID.String() + } + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventAgentLinkCreated, + Payload: payload, + }) + } } // --- Update --- @@ -261,6 +281,28 @@ func (m *AgentLinksMethods) handleUpdate(_ context.Context, client *gateway.Clie m.emitTeamCacheInvalidate() client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) + + // Emit agent_link.updated event + if m.msgBus != nil { + updatedLink, linkErr := m.linkStore.GetLink(ctx, linkID) + if linkErr == nil && updatedLink != nil { + changes := make([]string, 0, len(updates)) + for k := range updates { + changes = append(changes, k) + } + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventAgentLinkUpdated, + Payload: protocol.AgentLinkUpdatedPayload{ + LinkID: linkID.String(), + SourceAgentKey: updatedLink.SourceAgentKey, + TargetAgentKey: updatedLink.TargetAgentKey, + Direction: updatedLink.Direction, + Status: updatedLink.Status, + Changes: changes, + }, + }) + } + } } // --- Delete --- @@ -309,6 +351,18 @@ func (m *AgentLinksMethods) handleDelete(_ context.Context, client *gateway.Clie m.emitTeamCacheInvalidate() client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) + + // Emit agent_link.deleted event + if m.msgBus != nil && link != nil { + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventAgentLinkDeleted, + Payload: protocol.AgentLinkDeletedPayload{ + LinkID: linkID.String(), + SourceAgentKey: link.SourceAgentKey, + TargetAgentKey: link.TargetAgentKey, + }, + }) + } } // emitTeamCacheInvalidate broadcasts a cache invalidation event for team data. diff --git a/internal/gateway/methods/teams.go b/internal/gateway/methods/teams.go index 8a192ae0..faa222fd 100644 --- a/internal/gateway/methods/teams.go +++ b/internal/gateway/methods/teams.go @@ -167,4 +167,18 @@ func (m *TeamsMethods) handleCreate(_ context.Context, client *gateway.Client, r client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{ "team": team, })) + + // Emit team.created event + if m.msgBus != nil { + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventTeamCreated, + Payload: protocol.TeamCreatedPayload{ + TeamID: team.ID.String(), + TeamName: params.Name, + LeadAgentKey: leadAgent.AgentKey, + LeadDisplayName: leadAgent.DisplayName, + MemberCount: len(memberAgents) + 1, + }, + }) + } } diff --git a/internal/gateway/methods/teams_crud.go b/internal/gateway/methods/teams_crud.go index da278a7b..edfb12cb 100644 --- a/internal/gateway/methods/teams_crud.go +++ b/internal/gateway/methods/teams_crud.go @@ -6,6 +6,7 @@ import ( "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" @@ -90,7 +91,8 @@ func (m *TeamsMethods) handleDelete(_ context.Context, client *gateway.Client, r ctx := context.Background() - // Fetch members before deleting for cache invalidation + // Fetch team and members before deleting for event + cache invalidation + team, _ := m.teamStore.GetTeam(ctx, teamID) members, _ := m.teamStore.ListMembers(ctx, teamID) if err := m.teamStore.DeleteTeam(ctx, teamID); err != nil { @@ -106,6 +108,17 @@ func (m *TeamsMethods) handleDelete(_ context.Context, client *gateway.Client, r } client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) + + // Emit team.deleted event + if m.msgBus != nil && team != nil { + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventTeamDeleted, + Payload: protocol.TeamDeletedPayload{ + TeamID: teamID.String(), + TeamName: team.Name, + }, + }) + } } // --- Task List (admin view) --- @@ -183,7 +196,8 @@ func (m *TeamsMethods) handleUpdate(_ context.Context, client *gateway.Client, r ctx := context.Background() // Validate team exists - if _, err := m.teamStore.GetTeam(ctx, teamID); err != nil { + team, err := m.teamStore.GetTeam(ctx, teamID) + if err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "team not found: "+err.Error())) return } @@ -212,6 +226,22 @@ func (m *TeamsMethods) handleUpdate(_ context.Context, client *gateway.Client, r m.invalidateTeamCaches(ctx, teamID) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) + + // Emit team.updated event + if m.msgBus != nil { + changes := make([]string, 0, len(updates)) + for k := range updates { + changes = append(changes, k) + } + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventTeamUpdated, + Payload: protocol.TeamUpdatedPayload{ + TeamID: teamID.String(), + TeamName: team.Name, + Changes: changes, + }, + }) + } } // --- Known Users --- diff --git a/internal/gateway/methods/teams_members.go b/internal/gateway/methods/teams_members.go index 7697fcb0..67166afe 100644 --- a/internal/gateway/methods/teams_members.go +++ b/internal/gateway/methods/teams_members.go @@ -7,6 +7,7 @@ import ( "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" @@ -81,6 +82,21 @@ func (m *TeamsMethods) handleAddMember(_ context.Context, client *gateway.Client m.invalidateTeamCaches(ctx, teamID) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) + + // Emit team.member.added event + if m.msgBus != nil { + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventTeamMemberAdded, + Payload: protocol.TeamMemberAddedPayload{ + TeamID: teamID.String(), + TeamName: team.Name, + AgentID: ag.ID.String(), + AgentKey: ag.AgentKey, + DisplayName: ag.DisplayName, + Role: store.TeamRoleMember, + }, + }) + } } // --- Remove Member --- @@ -130,6 +146,9 @@ func (m *TeamsMethods) handleRemoveMember(_ context.Context, client *gateway.Cli return } + // Fetch agent info before removal for event emission + removedAgent, _ := m.agentStore.GetByID(ctx, agentID) + // Remove member if err := m.teamStore.RemoveMember(ctx, teamID, agentID); err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, "failed to remove member: "+err.Error())) @@ -153,6 +172,20 @@ func (m *TeamsMethods) handleRemoveMember(_ context.Context, client *gateway.Cli } client.SendResponse(protocol.NewOKResponse(req.ID, map[string]interface{}{"ok": true})) + + // Emit team.member.removed event + if m.msgBus != nil && removedAgent != nil { + m.msgBus.Broadcast(bus.Event{ + Name: protocol.EventTeamMemberRemoved, + Payload: protocol.TeamMemberRemovedPayload{ + TeamID: teamID.String(), + TeamName: team.Name, + AgentID: removedAgent.ID.String(), + AgentKey: removedAgent.AgentKey, + DisplayName: removedAgent.DisplayName, + }, + }) + } } // autoCreateTeamLinks creates outbound agent_links from lead to each member. diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index bb5bb0d9..da5e1806 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -21,6 +21,7 @@ type DelegationTask struct { SourceAgentID uuid.UUID `json:"source_agent_id"` SourceAgentKey string `json:"source_agent_key"` TargetAgentID uuid.UUID `json:"target_agent_id"` + SourceDisplayName string `json:"-"` TargetAgentKey string `json:"target_agent_key"` TargetDisplayName string `json:"-"` UserID string `json:"user_id"` @@ -72,6 +73,12 @@ type DelegateRunRequest struct { Stream bool ExtraSystemPrompt string MaxIterations int // per-delegation override (0 = use agent default) + + // Delegation context (bridged to agent.RunRequest for event enrichment) + DelegationID string + TeamID string + TeamTaskID string + ParentAgentID string } // DelegateRunResult is the result from AgentRunFunc. @@ -95,6 +102,7 @@ type DelegateArtifacts struct { // included in the final announce so the lead has all results in one message. type DelegateResultSummary struct { AgentKey string + DisplayName string // target agent display name Content string HasMedia bool Deliverables []string // actual content from tool outputs diff --git a/internal/tools/delegate_async.go b/internal/tools/delegate_async.go index 878a1123..7f6de919 100644 --- a/internal/tools/delegate_async.go +++ b/internal/tools/delegate_async.go @@ -10,6 +10,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/tracing" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) // DelegateAsync spawns a delegation in the background and announces the result back. @@ -31,7 +32,7 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) dm.injectDependencyResults(ctx, &opts) message := buildDelegateMessage(opts) - dm.emitEvent("delegation.started", task) + dm.emitDelegationEvent(protocol.EventDelegationStarted, task) slog.Info("delegation started (async)", "id", task.ID, "target", opts.TargetAgentKey) runReq := dm.buildRunRequest(task, message) @@ -88,20 +89,44 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) arts.Media = result.MediaPaths arts.Results = []DelegateResultSummary{{ AgentKey: task.TargetAgentKey, + DisplayName: task.TargetDisplayName, Content: result.Content, HasMedia: len(result.MediaPaths) > 0, Deliverables: result.Deliverables, }} } else if runErr != nil { arts.Results = []DelegateResultSummary{{ - AgentKey: task.TargetAgentKey, - Content: fmt.Sprintf("[failed] %s", runErr.Error()), + AgentKey: task.TargetAgentKey, + DisplayName: task.TargetDisplayName, + Content: fmt.Sprintf("[failed] %s", runErr.Error()), }} } if task.TeamTaskID != uuid.Nil { arts.CompletedTaskIDs = []string{task.TeamTaskID.String()} } dm.accumulateArtifacts(task.SourceAgentID, arts) + + // Emit accumulated event so WS clients know this delegation finished + // but results are being held until siblings complete. + if dm.msgBus != nil { + dm.msgBus.Broadcast(bus.Event{ + Name: protocol.EventDelegationAccumulated, + Payload: protocol.DelegationAccumulatedPayload{ + DelegationID: task.ID, + SourceAgentID: task.SourceAgentID.String(), + SourceAgentKey: task.SourceAgentKey, + TargetAgentKey: task.TargetAgentKey, + TargetDisplayName: task.TargetDisplayName, + UserID: task.UserID, + Channel: task.OriginChannel, + ChatID: task.OriginChatID, + TeamID: func() string { if task.TeamID != uuid.Nil { return task.TeamID.String() }; return "" }(), + TeamTaskID: func() string { if task.TeamTaskID != uuid.Nil { return task.TeamTaskID.String() }; return "" }(), + SiblingsRemaining: siblingCount, + ElapsedMS: int(time.Since(task.CreatedAt).Milliseconds()), + }, + }) + } slog.Info("delegation announce suppressed (siblings still running)", "id", task.ID, "target", task.TargetAgentKey, "siblings", siblingCount) } else { @@ -114,6 +139,7 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) artifacts.Media = append(artifacts.Media, result.MediaPaths...) artifacts.Results = append(artifacts.Results, DelegateResultSummary{ AgentKey: task.TargetAgentKey, + DisplayName: task.TargetDisplayName, Content: result.Content, HasMedia: len(result.MediaPaths) > 0, Deliverables: result.Deliverables, @@ -135,6 +161,41 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) if task.OriginLocalKey != "" { announceMeta["origin_local_key"] = task.OriginLocalKey } + // Emit announce event so WS clients know all results are being sent to lead. + hasMedia := len(artifacts.Media) > 0 + var announceSummaries []protocol.DelegationAnnounceResultSummary + for _, r := range artifacts.Results { + preview := r.Content + if len(preview) > 200 { + preview = preview[:200] + "..." + } + announceSummaries = append(announceSummaries, protocol.DelegationAnnounceResultSummary{ + AgentKey: r.AgentKey, + DisplayName: r.DisplayName, + HasMedia: r.HasMedia, + ContentPreview: preview, + }) + if r.HasMedia { + hasMedia = true + } + } + dm.msgBus.Broadcast(bus.Event{ + Name: protocol.EventDelegationAnnounce, + Payload: protocol.DelegationAnnouncePayload{ + SourceAgentID: task.SourceAgentID.String(), + SourceAgentKey: task.SourceAgentKey, + SourceDisplayName: task.SourceDisplayName, + UserID: task.UserID, + Channel: task.OriginChannel, + ChatID: task.OriginChatID, + TeamID: func() string { if task.TeamID != uuid.Nil { return task.TeamID.String() }; return "" }(), + Results: announceSummaries, + CompletedTaskIDs: artifacts.CompletedTaskIDs, + TotalElapsedMS: int(elapsed.Milliseconds()), + HasMedia: hasMedia, + }, + }) + announceMsg := bus.InboundMessage{ Channel: "system", SenderID: fmt.Sprintf("delegate:%s", task.ID), @@ -150,17 +211,17 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) if runErr != nil { task.Status = "failed" - dm.emitEvent("delegation.failed", task) + dm.emitDelegationEventWithError(task, runErr) dm.saveDelegationHistory(task, "", runErr, duration) } else { // Apply quality gates before marking completed. if result, runErr = dm.applyQualityGates(taskCtx, task, opts, result); runErr != nil { task.Status = "failed" - dm.emitEvent("delegation.failed", task) + dm.emitDelegationEventWithError(task, runErr) dm.saveDelegationHistory(task, "", runErr, duration) } else { task.Status = "completed" - dm.emitEvent("delegation.completed", task) + dm.emitDelegationEvent(protocol.EventDelegationCompleted, task) dm.trackCompleted(task) resultContent := "" var deliverables []string diff --git a/internal/tools/delegate_events.go b/internal/tools/delegate_events.go index 8c46819d..76ae224c 100644 --- a/internal/tools/delegate_events.go +++ b/internal/tools/delegate_events.go @@ -4,32 +4,76 @@ import ( "fmt" "time" + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) -func (dm *DelegateManager) emitEvent(name string, task *DelegationTask) { +// emitDelegationEvent broadcasts a delegation lifecycle event with a typed payload. +func (dm *DelegateManager) emitDelegationEvent(name string, task *DelegationTask) { if dm.msgBus == nil { return } - payload := map[string]string{ - "delegation_id": task.ID, - "source_agent": task.SourceAgentID.String(), - "target_agent": task.TargetAgentKey, - "user_id": task.UserID, - "mode": task.Mode, - } - if task.TeamID.String() != "00000000-0000-0000-0000-000000000000" { - payload["team_id"] = task.TeamID.String() - } - if task.TeamTaskID.String() != "00000000-0000-0000-0000-000000000000" { - payload["team_task_id"] = task.TeamTaskID.String() - } dm.msgBus.Broadcast(bus.Event{ Name: name, + Payload: buildDelegationPayload(task), + }) +} + +// emitDelegationEventWithError broadcasts a delegation failed event including the error. +func (dm *DelegateManager) emitDelegationEventWithError(task *DelegationTask, err error) { + if dm.msgBus == nil { + return + } + payload := buildDelegationPayload(task) + payload.Status = "failed" + payload.Error = err.Error() + if task.CompletedAt != nil { + payload.ElapsedMS = int(task.CompletedAt.Sub(task.CreatedAt).Milliseconds()) + } else { + payload.ElapsedMS = int(time.Since(task.CreatedAt).Milliseconds()) + } + dm.msgBus.Broadcast(bus.Event{ + Name: protocol.EventDelegationFailed, Payload: payload, }) } +// buildDelegationPayload creates a DelegationEventPayload from a DelegationTask. +func buildDelegationPayload(task *DelegationTask) protocol.DelegationEventPayload { + taskPreview := task.Task + if len(taskPreview) > 200 { + taskPreview = taskPreview[:200] + "..." + } + payload := protocol.DelegationEventPayload{ + DelegationID: task.ID, + SourceAgentID: task.SourceAgentID.String(), + SourceAgentKey: task.SourceAgentKey, + SourceDisplayName: task.SourceDisplayName, + TargetAgentID: task.TargetAgentID.String(), + TargetAgentKey: task.TargetAgentKey, + TargetDisplayName: task.TargetDisplayName, + UserID: task.UserID, + Channel: task.OriginChannel, + ChatID: task.OriginChatID, + Mode: task.Mode, + Task: taskPreview, + Status: task.Status, + CreatedAt: task.CreatedAt.UTC().Format(time.RFC3339), + } + if task.TeamID != uuid.Nil { + payload.TeamID = task.TeamID.String() + } + if task.TeamTaskID != uuid.Nil { + payload.TeamTaskID = task.TeamTaskID.String() + } + if task.CompletedAt != nil { + payload.ElapsedMS = int(task.CompletedAt.Sub(task.CreatedAt).Milliseconds()) + } + return payload +} + func formatDelegateAnnounce(task *DelegationTask, artifacts *DelegateArtifacts, err error, elapsed time.Duration) string { if err != nil && len(artifacts.Results) == 0 { return fmt.Sprintf( diff --git a/internal/tools/delegate_policy.go b/internal/tools/delegate_policy.go index 71f701f7..2dd4554c 100644 --- a/internal/tools/delegate_policy.go +++ b/internal/tools/delegate_policy.go @@ -6,7 +6,11 @@ import ( "fmt" "log/slog" + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/hooks" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) func checkUserPermission(settings json.RawMessage, userID string) error { @@ -204,6 +208,26 @@ func (dm *DelegateManager) applyQualityGates( "type", gate.Type, "delegation", task.ID, "attempt", attempt+1, "max_retries", retries) + // Emit quality gate retry event for WS visibility. + if dm.msgBus != nil { + dm.msgBus.Broadcast(bus.Event{ + Name: protocol.EventQualityGateRetry, + Payload: protocol.QualityGateRetryPayload{ + DelegationID: task.ID, + TargetAgentKey: task.TargetAgentKey, + UserID: task.UserID, + Channel: task.OriginChannel, + ChatID: task.OriginChatID, + TeamID: func() string { if task.TeamID != uuid.Nil { return task.TeamID.String() }; return "" }(), + TeamTaskID: func() string { if task.TeamTaskID != uuid.Nil { return task.TeamTaskID.String() }; return "" }(), + GateType: string(gate.Type), + Attempt: attempt + 1, + MaxRetries: retries, + Feedback: hookResult.Feedback, + }, + }) + } + feedbackMsg := fmt.Sprintf( "[Quality Gate Feedback — Retry %d/%d]\n"+ "Your previous output did not pass quality review.\n\n"+ diff --git a/internal/tools/delegate_prep.go b/internal/tools/delegate_prep.go index 294255a4..a7ea72b5 100644 --- a/internal/tools/delegate_prep.go +++ b/internal/tools/delegate_prep.go @@ -12,6 +12,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/internal/tracing" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateOpts, mode string) (*DelegationTask, *store.AgentLinkData, error) { @@ -157,7 +158,8 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO task := &DelegationTask{ ID: delegationID, SourceAgentID: sourceAgentID, - SourceAgentKey: sourceAgent.AgentKey, + SourceAgentKey: sourceAgent.AgentKey, + SourceDisplayName: sourceAgent.DisplayName, TargetAgentID: targetAgent.ID, TargetAgentKey: opts.TargetAgentKey, TargetDisplayName: targetAgent.DisplayName, @@ -270,6 +272,33 @@ func (dm *DelegateManager) sendProgressNotification(task *DelegationTask) { "peer_kind": task.OriginPeerKind, }, }) + + // Emit WS progress event alongside the outbound channel message. + var progressItems []protocol.DelegationProgressItem + for _, t := range active { + item := protocol.DelegationProgressItem{ + DelegationID: t.ID, + TargetAgentKey: t.TargetAgentKey, + TargetDisplayName: t.TargetDisplayName, + ElapsedMS: int(time.Since(t.CreatedAt).Milliseconds()), + } + if t.TeamTaskID != uuid.Nil { + item.TeamTaskID = t.TeamTaskID.String() + } + progressItems = append(progressItems, item) + } + dm.msgBus.Broadcast(bus.Event{ + Name: protocol.EventDelegationProgress, + Payload: protocol.DelegationProgressPayload{ + SourceAgentID: task.SourceAgentID.String(), + SourceAgentKey: task.SourceAgentKey, + UserID: task.UserID, + Channel: task.OriginChannel, + ChatID: task.OriginChatID, + TeamID: func() string { if task.TeamID != uuid.Nil { return task.TeamID.String() }; return "" }(), + Active: progressItems, + }, + }) } func buildDelegateMessage(opts DelegateOpts) string { @@ -296,5 +325,9 @@ func (dm *DelegateManager) buildRunRequest(task *DelegationTask, message string) "- Do NOT use your persona name or self-references (e.g. do not say your name). Write factual, neutral content.\n" + "- Be concise and deliver actionable results.\n" + "- IMPORTANT: If the delegated task falls outside your expertise scope (as defined in your SOUL.md), politely refuse and explain that this task is not within your domain. Do NOT attempt tasks outside your scope.", + DelegationID: task.ID, + TeamID: func() string { if task.TeamID != uuid.Nil { return task.TeamID.String() }; return "" }(), + TeamTaskID: func() string { if task.TeamTaskID != uuid.Nil { return task.TeamTaskID.String() }; return "" }(), + ParentAgentID: task.SourceAgentKey, } } diff --git a/internal/tools/delegate_state.go b/internal/tools/delegate_state.go index 5a8bbd42..08adb577 100644 --- a/internal/tools/delegate_state.go +++ b/internal/tools/delegate_state.go @@ -10,6 +10,7 @@ import ( "github.com/google/uuid" "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) // CancelForOrigin cancels all active async delegations originating from a given channel + chatID. @@ -26,7 +27,7 @@ func (dm *DelegateManager) CancelForOrigin(channel, chatID string) int { now := time.Now() t.CompletedAt = &now dm.active.Delete(key) - dm.emitEvent("delegation.cancelled", t) + dm.emitDelegationEvent(protocol.EventDelegationCancelled, t) slog.Info("delegation cancelled by /stopall", "id", t.ID, "target", t.TargetAgentKey) count++ } @@ -49,7 +50,7 @@ func (dm *DelegateManager) Cancel(delegationID string) bool { now := time.Now() task.CompletedAt = &now dm.active.Delete(delegationID) - dm.emitEvent("delegation.cancelled", task) + dm.emitDelegationEvent(protocol.EventDelegationCancelled, task) slog.Info("delegation cancelled", "id", delegationID, "target", task.TargetAgentKey) return true } @@ -68,7 +69,7 @@ func (dm *DelegateManager) CancelByTeamTaskID(teamTaskID uuid.UUID) bool { now := time.Now() t.CompletedAt = &now dm.active.Delete(key) - dm.emitEvent("delegation.cancelled", t) + dm.emitDelegationEvent(protocol.EventDelegationCancelled, t) slog.Info("delegation cancelled by team task cancel", "delegation_id", t.ID, "team_task_id", teamTaskID, "target", t.TargetAgentKey) found = true diff --git a/internal/tools/delegate_sync.go b/internal/tools/delegate_sync.go index 80584eff..6ba9438f 100644 --- a/internal/tools/delegate_sync.go +++ b/internal/tools/delegate_sync.go @@ -10,6 +10,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/internal/tracing" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) // Delegate executes a synchronous delegation to another agent. @@ -28,7 +29,7 @@ func (dm *DelegateManager) Delegate(ctx context.Context, opts DelegateOpts) (*De dm.injectDependencyResults(ctx, &opts) message := buildDelegateMessage(opts) - dm.emitEvent("delegation.started", task) + dm.emitDelegationEvent(protocol.EventDelegationStarted, task) slog.Info("delegation started", "id", task.ID, "target", opts.TargetAgentKey, "mode", "sync") // Propagate parent trace ID so the delegate trace links back. @@ -45,7 +46,7 @@ func (dm *DelegateManager) Delegate(ctx context.Context, opts DelegateOpts) (*De duration := time.Since(startTime) if err != nil { task.Status = "failed" - dm.emitEvent("delegation.failed", task) + dm.emitDelegationEventWithError(task, err) dm.saveDelegationHistory(task, "", err, duration) return nil, fmt.Errorf("delegation to %q failed: %w", opts.TargetAgentKey, err) } @@ -53,13 +54,13 @@ func (dm *DelegateManager) Delegate(ctx context.Context, opts DelegateOpts) (*De // Apply quality gates before marking completed. if result, err = dm.applyQualityGates(delegateCtx, task, opts, result); err != nil { task.Status = "failed" - dm.emitEvent("delegation.failed", task) + dm.emitDelegationEventWithError(task, err) dm.saveDelegationHistory(task, "", err, duration) return nil, fmt.Errorf("delegation to %q failed quality gate: %w", opts.TargetAgentKey, err) } task.Status = "completed" - dm.emitEvent("delegation.completed", task) + dm.emitDelegationEvent(protocol.EventDelegationCompleted, task) dm.trackCompleted(task) dm.autoCompleteTeamTask(task, result.Content, result.Deliverables) dm.saveDelegationHistory(task, result.Content, nil, duration) diff --git a/internal/tools/team_message_tool.go b/internal/tools/team_message_tool.go index 4ebd2b95..fb6d7f42 100644 --- a/internal/tools/team_message_tool.go +++ b/internal/tools/team_message_tool.go @@ -118,11 +118,17 @@ func (t *TeamMessageTool) executeSend(ctx context.Context, args map[string]inter if len(preview) > 100 { preview = preview[:100] + "..." } - t.manager.broadcastTeamEvent(protocol.EventTeamMessageSent, map[string]string{ - "team_id": team.ID.String(), - "from": fromKey, - "to": toKey, - "preview": preview, + t.manager.broadcastTeamEvent(protocol.EventTeamMessageSent, protocol.TeamMessageEventPayload{ + TeamID: team.ID.String(), + FromAgentKey: fromKey, + FromDisplayName: t.manager.agentDisplayName(ctx, fromKey), + ToAgentKey: toKey, + ToDisplayName: t.manager.agentDisplayName(ctx, toKey), + MessageType: string(store.TeamMessageTypeChat), + Preview: preview, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), }) return NewResult(fmt.Sprintf("Message sent to %s.", toKey)) @@ -167,11 +173,16 @@ func (t *TeamMessageTool) executeBroadcast(ctx context.Context, args map[string] if len(preview) > 100 { preview = preview[:100] + "..." } - t.manager.broadcastTeamEvent(protocol.EventTeamMessageSent, map[string]string{ - "team_id": team.ID.String(), - "from": fromKey, - "to": "broadcast", - "preview": preview, + t.manager.broadcastTeamEvent(protocol.EventTeamMessageSent, protocol.TeamMessageEventPayload{ + TeamID: team.ID.String(), + FromAgentKey: fromKey, + FromDisplayName: t.manager.agentDisplayName(ctx, fromKey), + ToAgentKey: "broadcast", + MessageType: string(store.TeamMessageTypeBroadcast), + Preview: preview, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), }) return NewResult(fmt.Sprintf("Broadcast sent to all teammates.")) diff --git a/internal/tools/team_tasks_tool.go b/internal/tools/team_tasks_tool.go index bf512fcd..f55055c6 100644 --- a/internal/tools/team_tasks_tool.go +++ b/internal/tools/team_tasks_tool.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "time" "github.com/google/uuid" @@ -272,11 +273,15 @@ func (t *TeamTasksTool) executeCreate(ctx context.Context, args map[string]inter return ErrorResult("failed to create task: " + err.Error()) } - t.manager.broadcastTeamEvent(protocol.EventTeamTaskCreated, map[string]string{ - "team_id": team.ID.String(), - "task_id": task.ID.String(), - "subject": subject, - "status": status, + t.manager.broadcastTeamEvent(protocol.EventTeamTaskCreated, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: task.ID.String(), + Subject: subject, + Status: status, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: task.CreatedAt.UTC().Format("2006-01-02T15:04:05Z"), }) return NewResult(fmt.Sprintf("Task created: %s (id=%s, status=%s)", subject, task.ID, status)) @@ -301,6 +306,19 @@ func (t *TeamTasksTool) executeClaim(ctx context.Context, args map[string]interf return ErrorResult("failed to claim task: " + err.Error()) } + ownerKey := t.manager.agentKeyFromID(ctx, agentID) + t.manager.broadcastTeamEvent(protocol.EventTeamTaskClaimed, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: taskIDStr, + Status: store.TeamTaskStatusInProgress, + OwnerAgentKey: ownerKey, + OwnerDisplayName: t.manager.agentDisplayName(ctx, ownerKey), + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), + }) + return NewResult(fmt.Sprintf("Task %s claimed successfully. It is now in progress.", taskIDStr)) } @@ -338,9 +356,17 @@ func (t *TeamTasksTool) executeComplete(ctx context.Context, args map[string]int return ErrorResult("failed to complete task: " + err.Error()) } - t.manager.broadcastTeamEvent(protocol.EventTeamTaskCompleted, map[string]string{ - "team_id": team.ID.String(), - "task_id": taskIDStr, + ownerKey := t.manager.agentKeyFromID(ctx, agentID) + t.manager.broadcastTeamEvent(protocol.EventTeamTaskCompleted, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: taskIDStr, + Status: store.TeamTaskStatusCompleted, + OwnerAgentKey: ownerKey, + OwnerDisplayName: t.manager.agentDisplayName(ctx, ownerKey), + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), }) return NewResult(fmt.Sprintf("Task %s completed. Dependent tasks have been unblocked.", taskIDStr)) @@ -381,9 +407,15 @@ func (t *TeamTasksTool) executeCancel(ctx context.Context, args map[string]inter t.manager.delegateMgr.CancelByTeamTaskID(taskID) } - t.manager.broadcastTeamEvent(protocol.EventTeamTaskCompleted, map[string]string{ - "team_id": team.ID.String(), - "task_id": taskIDStr, + t.manager.broadcastTeamEvent(protocol.EventTeamTaskCancelled, protocol.TeamTaskEventPayload{ + TeamID: team.ID.String(), + TaskID: taskIDStr, + Status: "cancelled", + Reason: reason, + UserID: store.UserIDFromContext(ctx), + Channel: ToolChannelFromCtx(ctx), + ChatID: ToolChatIDFromCtx(ctx), + Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"), }) return NewResult(fmt.Sprintf("Task %s cancelled. Any running delegation has been stopped and dependent tasks unblocked.", taskIDStr)) diff --git a/internal/tools/team_tool_manager.go b/internal/tools/team_tool_manager.go index 31bf924f..3f2a8e6c 100644 --- a/internal/tools/team_tool_manager.go +++ b/internal/tools/team_tool_manager.go @@ -113,7 +113,7 @@ func (m *TeamToolManager) agentKeyFromID(ctx context.Context, id uuid.UUID) stri } // broadcastTeamEvent sends a real-time event via the message bus for team activity visibility. -func (m *TeamToolManager) broadcastTeamEvent(name string, payload map[string]string) { +func (m *TeamToolManager) broadcastTeamEvent(name string, payload interface{}) { if m.msgBus == nil { return } @@ -122,3 +122,12 @@ func (m *TeamToolManager) broadcastTeamEvent(name string, payload map[string]str Payload: payload, }) } + +// agentDisplayName returns the display name for an agent key, falling back to empty string. +func (m *TeamToolManager) agentDisplayName(ctx context.Context, key string) string { + ag, err := m.agentStore.GetByKey(ctx, key) + if err != nil || ag.DisplayName == "" { + return "" + } + return ag.DisplayName +} diff --git a/pkg/protocol/events.go b/pkg/protocol/events.go index 76d58d0d..4eea797c 100644 --- a/pkg/protocol/events.go +++ b/pkg/protocol/events.go @@ -33,6 +33,30 @@ const ( EventDelegationStarted = "delegation.started" EventDelegationCompleted = "delegation.completed" + // Delegation lifecycle events. + EventDelegationFailed = "delegation.failed" + EventDelegationCancelled = "delegation.cancelled" + EventDelegationProgress = "delegation.progress" + EventDelegationAccumulated = "delegation.accumulated" + EventDelegationAnnounce = "delegation.announce" + EventQualityGateRetry = "delegation.quality_gate.retry" + + // Team task lifecycle events. + EventTeamTaskClaimed = "team.task.claimed" + EventTeamTaskCancelled = "team.task.cancelled" + + // Team CRUD events (admin operations). + EventTeamCreated = "team.created" + EventTeamUpdated = "team.updated" + EventTeamDeleted = "team.deleted" + EventTeamMemberAdded = "team.member.added" + EventTeamMemberRemoved = "team.member.removed" + + // Agent link events (admin operations). + EventAgentLinkCreated = "agent_link.created" + EventAgentLinkUpdated = "agent_link.updated" + EventAgentLinkDeleted = "agent_link.deleted" + // Cache invalidation events (internal, not forwarded to WS clients). EventCacheInvalidate = "cache.invalidate" diff --git a/pkg/protocol/team_events.go b/pkg/protocol/team_events.go new file mode 100644 index 00000000..56c064d3 --- /dev/null +++ b/pkg/protocol/team_events.go @@ -0,0 +1,206 @@ +package protocol + +// DelegationEventPayload is the typed payload for delegation lifecycle events +// (started, completed, failed, cancelled). +type DelegationEventPayload struct { + DelegationID string `json:"delegation_id"` + SourceAgentID string `json:"source_agent_id"` + SourceAgentKey string `json:"source_agent_key"` + SourceDisplayName string `json:"source_display_name,omitempty"` + TargetAgentID string `json:"target_agent_id"` + TargetAgentKey string `json:"target_agent_key"` + TargetDisplayName string `json:"target_display_name,omitempty"` + UserID string `json:"user_id"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Mode string `json:"mode"` + Task string `json:"task,omitempty"` + TeamID string `json:"team_id,omitempty"` + TeamTaskID string `json:"team_task_id,omitempty"` + Status string `json:"status,omitempty"` + ElapsedMS int `json:"elapsed_ms,omitempty"` + Error string `json:"error,omitempty"` + CreatedAt string `json:"created_at"` +} + +// DelegationProgressItem describes one active delegation in a progress report. +type DelegationProgressItem struct { + DelegationID string `json:"delegation_id"` + TargetAgentKey string `json:"target_agent_key"` + TargetDisplayName string `json:"target_display_name,omitempty"` + ElapsedMS int `json:"elapsed_ms"` + TeamTaskID string `json:"team_task_id,omitempty"` +} + +// DelegationProgressPayload is emitted periodically for async delegations. +type DelegationProgressPayload struct { + SourceAgentID string `json:"source_agent_id"` + SourceAgentKey string `json:"source_agent_key"` + UserID string `json:"user_id"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + TeamID string `json:"team_id,omitempty"` + Active []DelegationProgressItem `json:"active_delegations"` +} + +// DelegationAccumulatedPayload is emitted when an async delegation completes +// but siblings are still running (result accumulated, not yet announced). +type DelegationAccumulatedPayload struct { + DelegationID string `json:"delegation_id"` + SourceAgentID string `json:"source_agent_id"` + SourceAgentKey string `json:"source_agent_key"` + TargetAgentKey string `json:"target_agent_key"` + TargetDisplayName string `json:"target_display_name,omitempty"` + UserID string `json:"user_id"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + TeamID string `json:"team_id,omitempty"` + TeamTaskID string `json:"team_task_id,omitempty"` + SiblingsRemaining int `json:"siblings_remaining"` + ElapsedMS int `json:"elapsed_ms,omitempty"` +} + +// DelegationAnnounceResultSummary describes one delegation result in the announce payload. +type DelegationAnnounceResultSummary struct { + AgentKey string `json:"agent_key"` + DisplayName string `json:"display_name,omitempty"` + HasMedia bool `json:"has_media"` + ContentPreview string `json:"content_preview,omitempty"` +} + +// DelegationAnnouncePayload is emitted when the last sibling delegation completes +// and all accumulated results are announced back to the lead agent. +type DelegationAnnouncePayload struct { + SourceAgentID string `json:"source_agent_id"` + SourceAgentKey string `json:"source_agent_key"` + SourceDisplayName string `json:"source_display_name,omitempty"` + UserID string `json:"user_id"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + TeamID string `json:"team_id,omitempty"` + Results []DelegationAnnounceResultSummary `json:"results"` + CompletedTaskIDs []string `json:"completed_task_ids,omitempty"` + TotalElapsedMS int `json:"total_elapsed_ms"` + HasMedia bool `json:"has_media"` +} + +// QualityGateRetryPayload is emitted when a quality gate rejects a delegation +// result and triggers a retry. +type QualityGateRetryPayload struct { + DelegationID string `json:"delegation_id"` + TargetAgentKey string `json:"target_agent_key"` + UserID string `json:"user_id"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + TeamID string `json:"team_id,omitempty"` + TeamTaskID string `json:"team_task_id,omitempty"` + GateType string `json:"gate_type"` + Attempt int `json:"attempt"` + MaxRetries int `json:"max_retries"` + Feedback string `json:"feedback,omitempty"` +} + +// TeamTaskEventPayload is the typed payload for team task lifecycle events +// (created, claimed, completed, cancelled). +type TeamTaskEventPayload struct { + TeamID string `json:"team_id"` + TaskID string `json:"task_id"` + Subject string `json:"subject,omitempty"` + Status string `json:"status"` + OwnerAgentKey string `json:"owner_agent_key,omitempty"` + OwnerDisplayName string `json:"owner_display_name,omitempty"` + Reason string `json:"reason,omitempty"` + UserID string `json:"user_id"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Timestamp string `json:"timestamp"` +} + +// TeamMessageEventPayload is the typed payload for team.message.sent events. +type TeamMessageEventPayload struct { + TeamID string `json:"team_id"` + FromAgentKey string `json:"from_agent_key"` + FromDisplayName string `json:"from_display_name,omitempty"` + ToAgentKey string `json:"to_agent_key"` + ToDisplayName string `json:"to_display_name,omitempty"` + MessageType string `json:"message_type"` + Preview string `json:"preview"` + TaskID string `json:"task_id,omitempty"` + UserID string `json:"user_id"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` +} + +// --- Team CRUD event payloads --- + +// TeamCreatedPayload is emitted when a new team is created via RPC. +type TeamCreatedPayload struct { + TeamID string `json:"team_id"` + TeamName string `json:"team_name"` + LeadAgentKey string `json:"lead_agent_key"` + LeadDisplayName string `json:"lead_display_name,omitempty"` + MemberCount int `json:"member_count"` +} + +// TeamUpdatedPayload is emitted when team settings are updated via RPC. +type TeamUpdatedPayload struct { + TeamID string `json:"team_id"` + TeamName string `json:"team_name"` + Changes []string `json:"changes"` +} + +// TeamDeletedPayload is emitted when a team is deleted via RPC. +type TeamDeletedPayload struct { + TeamID string `json:"team_id"` + TeamName string `json:"team_name"` +} + +// TeamMemberAddedPayload is emitted when a member is added to a team. +type TeamMemberAddedPayload struct { + TeamID string `json:"team_id"` + TeamName string `json:"team_name"` + AgentID string `json:"agent_id"` + AgentKey string `json:"agent_key"` + DisplayName string `json:"display_name,omitempty"` + Role string `json:"role"` +} + +// TeamMemberRemovedPayload is emitted when a member is removed from a team. +type TeamMemberRemovedPayload struct { + TeamID string `json:"team_id"` + TeamName string `json:"team_name"` + AgentID string `json:"agent_id"` + AgentKey string `json:"agent_key"` + DisplayName string `json:"display_name,omitempty"` +} + +// --- Agent link event payloads --- + +// AgentLinkCreatedPayload is emitted when a new agent link is created. +type AgentLinkCreatedPayload struct { + LinkID string `json:"link_id"` + SourceAgentID string `json:"source_agent_id"` + SourceAgentKey string `json:"source_agent_key"` + TargetAgentID string `json:"target_agent_id"` + TargetAgentKey string `json:"target_agent_key"` + Direction string `json:"direction"` + TeamID string `json:"team_id,omitempty"` + Status string `json:"status"` +} + +// AgentLinkUpdatedPayload is emitted when an agent link is updated. +type AgentLinkUpdatedPayload struct { + LinkID string `json:"link_id"` + SourceAgentKey string `json:"source_agent_key"` + TargetAgentKey string `json:"target_agent_key"` + Direction string `json:"direction,omitempty"` + Status string `json:"status,omitempty"` + Changes []string `json:"changes"` +} + +// AgentLinkDeletedPayload is emitted when an agent link is deleted. +type AgentLinkDeletedPayload struct { + LinkID string `json:"link_id"` + SourceAgentKey string `json:"source_agent_key"` + TargetAgentKey string `json:"target_agent_key"` +}