mirror of
https://github.com/tiennm99/goclaw.git
synced 2026-10-14 04:16:06 +00:00
feat(ws): comprehensive team & delegation WS events with typed payloads
Add 16 new event constants and typed payload structs for full WS visibility into team agent operations. Enrich AgentEvent with delegation and routing context (delegationId, teamId, parentAgentId, userId, channel, chatId). Emit thinking/chunk events for non-streaming runs so delegate member agents also produce WS events. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
viettranxandClaude Opus 4.6 committed
1 parent
936e958699
commit
61da687ad1
20 files changed
+1213
-62
No files matched your search
@@ -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
|
||||
|
||||
@@ -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": "<event_name>", "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`.
|
||||
+54
-10
@@ -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,
|
||||
|
||||
@@ -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 <agentID>")
|
||||
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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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 ---
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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"+
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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."))
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
Reference in new issue
Block a user